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

[Zac Medico]

commit:     85ac23b7c0c58cef72d22281d66d086521c01e3e
Author:     Zac Medico <zmedico <AT> gentoo <DOT> org>
AuthorDate: Sun May  6 11:05:03 2018 +0000
Commit:     Zac Medico <zmedico <AT> gentoo <DOT> org>
CommitDate: Sun May  6 11:41:45 2018 +0000
URL:        https://gitweb.gentoo.org/proj/portage.git/commit/?id=85ac23b7

asyncio: add _wrap_loop helper (bug 654390)

In order to deal with asyncio event loop compatibility issues, add
a _wrap_loop helper. For example, since python3.4 does not have the
AbstractEventLoop.create_future() method, this helper function can
be used to add a wrapper that implements the create_future method
for python3.4.

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

 pym/portage/dbapi/porttree.py                         | 12 ++++++------
 .../tests/util/futures/asyncio/test_child_watcher.py  |  2 +-
 .../util/futures/asyncio/test_event_loop_in_fork.py   |  8 ++++----
 .../tests/util/futures/asyncio/test_pipe_closed.py    |  4 ++--
 .../util/futures/asyncio/test_run_until_complete.py   |  2 +-
 .../util/futures/asyncio/test_subprocess_exec.py      |  4 ++--
 pym/portage/util/futures/_asyncio/__init__.py         | 19 ++++++++++++++++++-
 pym/portage/util/futures/_asyncio/tasks.py            |  7 +++++--
 pym/portage/util/futures/executor/fork.py             |  4 ++--
 pym/portage/util/futures/iter_completed.py            |  7 +++----
 pym/portage/util/futures/retry.py                     |  3 +--
 11 files changed, 45 insertions(+), 27 deletions(-)

diff --git a/pym/portage/dbapi/porttree.py b/pym/portage/dbapi/porttree.py
index 801b5658a..3e36024ff 100644
--- a/pym/portage/dbapi/porttree.py
+++ b/pym/portage/dbapi/porttree.py
@@ -36,7 +36,7 @@ from portage import _encodings
 from portage import _unicode_encode
 from portage import OrderedDict
 from portage.util._eventloop.EventLoop import EventLoop
-from portage.util._eventloop.global_event_loop import global_event_loop
+from portage.util.futures import asyncio
 from portage.util.futures.iter_completed import iter_gather
 from _emerge.EbuildMetadataPhase import EbuildMetadataPhase
 
@@ -325,8 +325,8 @@ class portdbapi(dbapi):
 	@property
 	def _event_loop(self):
 		if portage._internal_caller:
-			# For internal portage usage, the global_event_loop is safe.
-			return global_event_loop()
+			# For internal portage usage, asyncio._wrap_loop() is safe.
+			return asyncio._wrap_loop()
 		else:
 			# For external API consumers, use a local EventLoop, since
 			# we don't want to assume that it's safe to override the
@@ -611,7 +611,7 @@ class portdbapi(dbapi):
 		# to simultaneous instantiation of multiple event loops here.
 		# Callers of this method certainly want the same event loop to
 		# be used for all calls.
-		loop = loop or global_event_loop()
+		loop = asyncio._wrap_loop(loop)
 		future = loop.create_future()
 		cache_me = False
 		if myrepo is not None:
@@ -751,7 +751,7 @@ class portdbapi(dbapi):
 			a set of alternative URIs.
 		@rtype: asyncio.Future (or compatible)
 		"""
-		loop = loop or global_event_loop()
+		loop = asyncio._wrap_loop(loop)
 		result = loop.create_future()
 
 		def aux_get_done(aux_get_future):
@@ -1419,7 +1419,7 @@ def _async_manifest_fetchlist(portdb, repo_config, cp, cpv_list=None,
 	@return: a Future resulting in a Mapping compatible with FetchlistDict
 	@rtype: asyncio.Future (or compatible)
 	"""
-	loop = loop or global_event_loop()
+	loop = asyncio._wrap_loop(loop)
 	result = loop.create_future()
 	cpv_list = (portdb.cp_list(cp, mytree=repo_config.location)
 		if cpv_list is None else cpv_list)

diff --git a/pym/portage/tests/util/futures/asyncio/test_child_watcher.py b/pym/portage/tests/util/futures/asyncio/test_child_watcher.py
index dca01be56..8ef497544 100644
--- a/pym/portage/tests/util/futures/asyncio/test_child_watcher.py
+++ b/pym/portage/tests/util/futures/asyncio/test_child_watcher.py
@@ -28,7 +28,7 @@ class ChildWatcherTestCase(TestCase):
 
 			args_tuple = ('hello', 'world')
 
-			loop = asyncio.get_event_loop()
+			loop = asyncio._wrap_loop()
 			future = loop.create_future()
 
 			def callback(pid, returncode, *args):

diff --git a/pym/portage/tests/util/futures/asyncio/test_event_loop_in_fork.py b/pym/portage/tests/util/futures/asyncio/test_event_loop_in_fork.py
index 7868d792a..19588bf3a 100644
--- a/pym/portage/tests/util/futures/asyncio/test_event_loop_in_fork.py
+++ b/pym/portage/tests/util/futures/asyncio/test_event_loop_in_fork.py
@@ -11,14 +11,14 @@ from portage.util.futures.unix_events import DefaultEventLoopPolicy
 
 def fork_main(parent_conn, child_conn):
 	parent_conn.close()
-	loop = asyncio.get_event_loop()
+	loop = asyncio._wrap_loop()
 	# This fails with python's default event loop policy,
 	# see https://bugs.python.org/issue22087.
-	loop.run_until_complete(asyncio.sleep(0.1))
+	loop.run_until_complete(asyncio.sleep(0.1, loop=loop))
 
 
 def async_main(fork_exitcode, loop=None):
-	loop = loop or asyncio.get_event_loop()
+	loop = asyncio._wrap_loop(loop)
 
 	# Since python2.7 does not support Process.sentinel, use Pipe to
 	# monitor for process exit.
@@ -48,7 +48,7 @@ class EventLoopInForkTestCase(TestCase):
 		if not isinstance(initial_policy, DefaultEventLoopPolicy):
 			asyncio.set_event_loop_policy(DefaultEventLoopPolicy())
 		try:
-			loop = asyncio.get_event_loop()
+			loop = asyncio._wrap_loop()
 			fork_exitcode = loop.create_future()
 			# Make async_main fork while the loop is running, which would
 			# trigger https://bugs.python.org/issue22087 with asyncio's

diff --git a/pym/portage/tests/util/futures/asyncio/test_pipe_closed.py b/pym/portage/tests/util/futures/asyncio/test_pipe_closed.py
index e63829888..c2b468064 100644
--- a/pym/portage/tests/util/futures/asyncio/test_pipe_closed.py
+++ b/pym/portage/tests/util/futures/asyncio/test_pipe_closed.py
@@ -58,7 +58,7 @@ class ReaderPipeClosedTestCase(_PipeClosedTestCase, TestCase):
 		if not isinstance(initial_policy, DefaultEventLoopPolicy):
 			asyncio.set_event_loop_policy(DefaultEventLoopPolicy())
 
-		loop = asyncio.get_event_loop()
+		loop = asyncio._wrap_loop()
 		read_end = os.fdopen(read_end, 'rb', 0)
 		write_end = os.fdopen(write_end, 'wb', 0)
 		try:
@@ -95,7 +95,7 @@ class WriterPipeClosedTestCase(_PipeClosedTestCase, TestCase):
 		if not isinstance(initial_policy, DefaultEventLoopPolicy):
 			asyncio.set_event_loop_policy(DefaultEventLoopPolicy())
 
-		loop = asyncio.get_event_loop()
+		loop = asyncio._wrap_loop()
 		read_end = os.fdopen(read_end, 'rb', 0)
 		write_end = os.fdopen(write_end, 'wb', 0)
 		try:

diff --git a/pym/portage/tests/util/futures/asyncio/test_run_until_complete.py b/pym/portage/tests/util/futures/asyncio/test_run_until_complete.py
index fc8f198ca..1a37e4922 100644
--- a/pym/portage/tests/util/futures/asyncio/test_run_until_complete.py
+++ b/pym/portage/tests/util/futures/asyncio/test_run_until_complete.py
@@ -13,7 +13,7 @@ class RunUntilCompleteTestCase(TestCase):
 			asyncio.set_event_loop_policy(DefaultEventLoopPolicy())
 
 		try:
-			loop = asyncio.get_event_loop()
+			loop = asyncio._wrap_loop()
 			f1 = loop.create_future()
 			f2 = loop.create_future()
 			f1.add_done_callback(f2.set_result)

diff --git a/pym/portage/tests/util/futures/asyncio/test_subprocess_exec.py b/pym/portage/tests/util/futures/asyncio/test_subprocess_exec.py
index 98983941d..8dc5fa7b9 100644
--- a/pym/portage/tests/util/futures/asyncio/test_subprocess_exec.py
+++ b/pym/portage/tests/util/futures/asyncio/test_subprocess_exec.py
@@ -23,7 +23,7 @@ def reader(input_file, loop=None):
 	@return: bytes
 	@rtype: asyncio.Future (or compatible)
 	"""
-	loop = loop or asyncio.get_event_loop()
+	loop = asyncio._wrap_loop(loop)
 	future = loop.create_future()
 	_Reader(future, input_file, loop)
 	return future
@@ -61,7 +61,7 @@ class SubprocessExecTestCase(TestCase):
 			asyncio.set_event_loop_policy(DefaultEventLoopPolicy())
 
 		try:
-			test(asyncio.get_event_loop())
+			test(asyncio._wrap_loop())
 		finally:
 			asyncio.set_event_loop_policy(initial_policy)
 

diff --git a/pym/portage/util/futures/_asyncio/__init__.py b/pym/portage/util/futures/_asyncio/__init__.py
index 9ae050874..e62de7a69 100644
--- a/pym/portage/util/futures/_asyncio/__init__.py
+++ b/pym/portage/util/futures/_asyncio/__init__.py
@@ -137,7 +137,7 @@ def sleep(delay, result=None, loop=None):
 	@rtype: asyncio.Future (or compatible)
 	@return: an instance of Future
 	"""
-	loop = loop or get_event_loop()
+	loop = _wrap_loop(loop)
 	future = loop.create_future()
 	handle = loop.call_later(delay, future.set_result, result)
 	def cancel_callback(future):
@@ -145,3 +145,20 @@ def sleep(delay, result=None, loop=None):
 			handle.cancel()
 	future.add_done_callback(cancel_callback)
 	return future
+
+
+def _wrap_loop(loop=None):
+	"""
+	In order to deal with asyncio event loop compatibility issues,
+	use this function to wrap the loop parameter for functions
+	that support it. For example, since python3.4 does not have the
+	AbstractEventLoop.create_future() method, this helper function
+	can be used to add a wrapper that implements the create_future
+	method for python3.4.
+
+	@type loop: asyncio.AbstractEventLoop (or compatible)
+	@param loop: event loop
+	@rtype: asyncio.AbstractEventLoop (or compatible)
+	@return: event loop
+	"""
+	return loop or get_event_loop()

diff --git a/pym/portage/util/futures/_asyncio/tasks.py b/pym/portage/util/futures/_asyncio/tasks.py
index 5f10d3c7b..b20765b7a 100644
--- a/pym/portage/util/futures/_asyncio/tasks.py
+++ b/pym/portage/util/futures/_asyncio/tasks.py
@@ -15,7 +15,10 @@ except ImportError:
 	FIRST_COMPLETED ='FIRST_COMPLETED'
 	FIRST_EXCEPTION = 'FIRST_EXCEPTION'
 
-
+import portage
+portage.proxy.lazyimport.lazyimport(globals(),
+	'portage.util.futures:asyncio',
+)
 from portage.util._eventloop.global_event_loop import (
 	global_event_loop as _global_event_loop,
 )
@@ -40,7 +43,7 @@ def wait(futures, loop=None, timeout=None, return_when=ALL_COMPLETED):
 	@return: tuple of (done, pending).
 	@rtype: asyncio.Future (or compatible)
 	"""
-	loop = loop or _global_event_loop()
+	loop = asyncio._wrap_loop(loop)
 	result_future = loop.create_future()
 	_Waiter(futures, timeout, return_when, result_future, loop)
 	return result_future

diff --git a/pym/portage/util/futures/executor/fork.py b/pym/portage/util/futures/executor/fork.py
index 276ed54f1..72844403c 100644
--- a/pym/portage/util/futures/executor/fork.py
+++ b/pym/portage/util/futures/executor/fork.py
@@ -13,7 +13,7 @@ import sys
 import traceback
 
 from portage.util._async.AsyncFunction import AsyncFunction
-from portage.util._eventloop.global_event_loop import global_event_loop
+from portage.util.futures import asyncio
 
 
 class ForkExecutor(object):
@@ -25,7 +25,7 @@ class ForkExecutor(object):
 	"""
 	def __init__(self, max_workers=None, loop=None):
 		self._max_workers = max_workers or multiprocessing.cpu_count()
-		self._loop = loop or global_event_loop()
+		self._loop = asyncio._wrap_loop(loop)
 		self._submit_queue = collections.deque()
 		self._running_tasks = {}
 		self._shutdown = False

diff --git a/pym/portage/util/futures/iter_completed.py b/pym/portage/util/futures/iter_completed.py
index 231b7e3ab..31b5e0c78 100644
--- a/pym/portage/util/futures/iter_completed.py
+++ b/pym/portage/util/futures/iter_completed.py
@@ -6,7 +6,6 @@ import multiprocessing
 
 from portage.util._async.AsyncTaskFuture import AsyncTaskFuture
 from portage.util._async.TaskScheduler import TaskScheduler
-from portage.util._eventloop.global_event_loop import global_event_loop
 from portage.util.futures import asyncio
 
 
@@ -30,7 +29,7 @@ def iter_completed(futures, max_jobs=None, max_load=None, loop=None):
 	@return: iterator of futures that are done
 	@rtype: iterator
 	"""
-	loop = loop or global_event_loop()
+	loop = asyncio._wrap_loop(loop)
 
 	for future_done_set in async_iter_completed(futures,
 		max_jobs=max_jobs, max_load=max_load, loop=loop):
@@ -60,7 +59,7 @@ def async_iter_completed(futures, max_jobs=None, max_load=None, loop=None):
 		input futures that are done
 	@rtype: iterator
 	"""
-	loop = loop or global_event_loop()
+	loop = asyncio._wrap_loop(loop)
 
 	max_jobs = max_jobs or multiprocessing.cpu_count()
 	max_load = max_load or multiprocessing.cpu_count()
@@ -133,7 +132,7 @@ def iter_gather(futures, max_jobs=None, max_load=None, loop=None):
 		same order that they were yielded from the input iterator
 	@rtype: asyncio.Future (or compatible)
 	"""
-	loop = loop or global_event_loop()
+	loop = asyncio._wrap_loop(loop)
 	result = loop.create_future()
 	futures_list = []
 

diff --git a/pym/portage/util/futures/retry.py b/pym/portage/util/futures/retry.py
index 82012d2f3..8a51669ff 100644
--- a/pym/portage/util/futures/retry.py
+++ b/pym/portage/util/futures/retry.py
@@ -9,7 +9,6 @@ __all__ = (
 import functools
 
 from portage.exception import PortageException
-from portage.util._eventloop.global_event_loop import global_event_loop
 from portage.util.futures import asyncio
 
 
@@ -67,7 +66,7 @@ def _retry(loop, try_max, try_timeout, overall_timeout, delay_func,
 	@return: func return value
 	@rtype: asyncio.Future (or compatible)
 	"""
-	loop = loop or global_event_loop()
+	loop = asyncio._wrap_loop(loop)
 	future = loop.create_future()
 	_Retry(future, loop, try_max, try_timeout, overall_timeout, delay_func,
 		reraise, functools.partial(func, *args, **kwargs))