[gentoo-commits] proj/portage:repoman commit in: pym/portage/util/futures/

[gentoo-commits] proj/portage:repoman commit in: pym/portage/util/futures/

[Brian Dolbec]

commit:     1b23a4134f0919e9541e7544213321efc51cac81
Author:     Brian Dolbec <dolsen <AT> gentoo <DOT> org>
AuthorDate: Mon Apr 25 01:09:54 2016 +0000
Commit:     Brian Dolbec <dolsen <AT> gentoo <DOT> org>
CommitDate: Mon Apr 25 15:28:53 2016 +0000
URL:        https://gitweb.gentoo.org/proj/portage.git/commit/?id=1b23a413

portage: Create a new extended futures class which adds standard data type access

This new class adds standard get, set functions along with optional defaults.
It also adds the capability to ignore InvalidStateErrors when trying to set the reslt more 
than once.

 pym/portage/util/futures/extendedfutures.py | 73 +++++++++++++++++++++++++++++
 1 file changed, 73 insertions(+)

diff --git a/pym/portage/util/futures/extendedfutures.py b/pym/portage/util/futures/extendedfutures.py
new file mode 100644
index 0000000..af384c7
--- /dev/null
+++ b/pym/portage/util/futures/extendedfutures.py
@@ -0,0 +1,73 @@
+# Copyright 2016 Gentoo Foundation
+# Distributed under the terms of the GNU General Public License v2
+#
+# This module provides an extended subset of the asyncio.futures.Futures
+# interface.
+
+from __future__ import unicode_literals
+
+__all__ = (
+	'CancelledError',
+	'ExtendedFuture',
+	'InvalidStateError',
+)
+
+from portage.util.futures.futures import (Future, InvalidStateError,
+	CancelledError)
+
+# Create our one time settable unset constant
+UNSET_CONST = Future()
+UNSET_CONST.set_result(object())
+
+
+class ExtendedFuture(Future):
+	'''Extended Future class adding convienince get and set operations with
+	default result capabilities for unset result().  It also adds pass
+	capability for duplicate set_result() calls.
+	'''
+
+	def __init__(self, default_result=UNSET_CONST.result()):
+		'''Class init
+
+		@param default_result: Optional data type/value to return in the event
+		                       of a result() call when result has not yet been
+		                       set.
+		'''
+		self.default_result = default_result
+		super(ExtendedFuture, self).__init__()
+		self.set = self.set_result
+
+	def set_result(self, data, ignore_InvalidState=False):
+		'''Set the Future's result to the data, optionally don't raise
+		an error for 'InvalidStateError' errors
+
+		@param ignore_exception: Boolean
+		'''
+		if ignore_InvalidState:
+			try:
+				super(ExtendedFuture, self).set_result(data)
+			except InvalidStateError:
+				pass
+		else:
+			super(ExtendedFuture, self).set_result(data)
+
+	def get(self, default=UNSET_CONST.result()):
+		'''Convienience function to wrap result() but adds an optional
+		default value to return rather than raise an InvalidStateError
+
+		@param default: Optional override for the classwide default_result
+		@returns: the result data or the default value, raisies an exception
+		          if result is unset and no default is defined.
+		'''
+		if default is not UNSET_CONST.result():
+			pass
+		elif self.default_result is not UNSET_CONST.result():
+			default = self.default_result
+		if default is not UNSET_CONST.result():
+			try:
+				data = super(ExtendedFuture, self).result()
+			except InvalidStateError:
+				data = default
+		else:
+			data = super(ExtendedFuture, self).result()
+		return data

[gentoo-commits] proj/portage:master commit in: pym/portage/util/futures/

[Brian Dolbec]

commit:     1b23a4134f0919e9541e7544213321efc51cac81
Author:     Brian Dolbec <dolsen <AT> gentoo <DOT> org>
AuthorDate: Mon Apr 25 01:09:54 2016 +0000
Commit:     Brian Dolbec <dolsen <AT> gentoo <DOT> org>
CommitDate: Mon Apr 25 15:28:53 2016 +0000
URL:        https://gitweb.gentoo.org/proj/portage.git/commit/?id=1b23a413

portage: Create a new extended futures class which adds standard data type access

This new class adds standard get, set functions along with optional defaults.
It also adds the capability to ignore InvalidStateErrors when trying to set the reslt more 
than once.

 pym/portage/util/futures/extendedfutures.py | 73 +++++++++++++++++++++++++++++
 1 file changed, 73 insertions(+)

diff --git a/pym/portage/util/futures/extendedfutures.py b/pym/portage/util/futures/extendedfutures.py
new file mode 100644
index 0000000..af384c7
--- /dev/null
+++ b/pym/portage/util/futures/extendedfutures.py
@@ -0,0 +1,73 @@
+# Copyright 2016 Gentoo Foundation
+# Distributed under the terms of the GNU General Public License v2
+#
+# This module provides an extended subset of the asyncio.futures.Futures
+# interface.
+
+from __future__ import unicode_literals
+
+__all__ = (
+	'CancelledError',
+	'ExtendedFuture',
+	'InvalidStateError',
+)
+
+from portage.util.futures.futures import (Future, InvalidStateError,
+	CancelledError)
+
+# Create our one time settable unset constant
+UNSET_CONST = Future()
+UNSET_CONST.set_result(object())
+
+
+class ExtendedFuture(Future):
+	'''Extended Future class adding convienince get and set operations with
+	default result capabilities for unset result().  It also adds pass
+	capability for duplicate set_result() calls.
+	'''
+
+	def __init__(self, default_result=UNSET_CONST.result()):
+		'''Class init
+
+		@param default_result: Optional data type/value to return in the event
+		                       of a result() call when result has not yet been
+		                       set.
+		'''
+		self.default_result = default_result
+		super(ExtendedFuture, self).__init__()
+		self.set = self.set_result
+
+	def set_result(self, data, ignore_InvalidState=False):
+		'''Set the Future's result to the data, optionally don't raise
+		an error for 'InvalidStateError' errors
+
+		@param ignore_exception: Boolean
+		'''
+		if ignore_InvalidState:
+			try:
+				super(ExtendedFuture, self).set_result(data)
+			except InvalidStateError:
+				pass
+		else:
+			super(ExtendedFuture, self).set_result(data)
+
+	def get(self, default=UNSET_CONST.result()):
+		'''Convienience function to wrap result() but adds an optional
+		default value to return rather than raise an InvalidStateError
+
+		@param default: Optional override for the classwide default_result
+		@returns: the result data or the default value, raisies an exception
+		          if result is unset and no default is defined.
+		'''
+		if default is not UNSET_CONST.result():
+			pass
+		elif self.default_result is not UNSET_CONST.result():
+			default = self.default_result
+		if default is not UNSET_CONST.result():
+			try:
+				data = super(ExtendedFuture, self).result()
+			except InvalidStateError:
+				data = default
+		else:
+			data = super(ExtendedFuture, self).result()
+		return data

[gentoo-commits] proj/portage:master commit in: pym/portage/util/futures/

[Zac Medico]

commit:     d90ff7c046b39de82558655375ea104cfa19b176
Author:     Zac Medico <zmedico <AT> gentoo <DOT> org>
AuthorDate: Sun Mar 26 20:08:54 2017 +0000
Commit:     Zac Medico <zmedico <AT> gentoo <DOT> org>
CommitDate: Sun Mar 26 20:12:04 2017 +0000
URL:        https://gitweb.gentoo.org/proj/portage.git/commit/?id=d90ff7c0

_EventLoopFuture: reduce indent of class body (whitespace only)

 pym/portage/util/futures/futures.py | 278 ++++++++++++++++++------------------
 1 file changed, 139 insertions(+), 139 deletions(-)

diff --git a/pym/portage/util/futures/futures.py b/pym/portage/util/futures/futures.py
index dd913a1e3..dcf593c01 100644
--- a/pym/portage/util/futures/futures.py
+++ b/pym/portage/util/futures/futures.py
@@ -42,148 +42,148 @@ _CANCELLED = 'CANCELLED'
 _FINISHED = 'FINISHED'
 
 class _EventLoopFuture(object):
+	"""
+	This class provides (a subset of) the asyncio.Future interface, for
+	use with the EventLoop class, because EventLoop is currently
+	missing some of the asyncio.AbstractEventLoop methods that
+	asyncio.Future requires.
+	"""
+
+	# Class variables serving as defaults for instance variables.
+	_state = _PENDING
+	_result = None
+	_exception = None
+	_loop = None
+
+	def __init__(self, loop=None):
+		"""Initialize the future.
+
+		The optional loop argument allows explicitly setting the event
+		loop object used by the future. If it's not provided, the future uses
+		the default event loop.
 		"""
-		This class provides (a subset of) the asyncio.Future interface, for
-		use with the EventLoop class, because EventLoop is currently
-		missing some of the asyncio.AbstractEventLoop methods that
-		asyncio.Future requires.
+		if loop is None:
+			self._loop = global_event_loop()
+		else:
+			self._loop = loop
+		self._callbacks = []
+
+	def cancel(self):
+		"""Cancel the future and schedule callbacks.
+
+		If the future is already done or cancelled, return False.  Otherwise,
+		change the future's state to cancelled, schedule the callbacks and
+		return True.
 		"""
+		if self._state != _PENDING:
+			return False
+		self._state = _CANCELLED
+		self._schedule_callbacks()
+		return True
 
-		# Class variables serving as defaults for instance variables.
-		_state = _PENDING
-		_result = None
-		_exception = None
-		_loop = None
-
-		def __init__(self, loop=None):
-			"""Initialize the future.
-
-			The optional loop argument allows explicitly setting the event
-			loop object used by the future. If it's not provided, the future uses
-			the default event loop.
-			"""
-			if loop is None:
-				self._loop = global_event_loop()
-			else:
-				self._loop = loop
-			self._callbacks = []
-
-		def cancel(self):
-			"""Cancel the future and schedule callbacks.
-
-			If the future is already done or cancelled, return False.  Otherwise,
-			change the future's state to cancelled, schedule the callbacks and
-			return True.
-			"""
-			if self._state != _PENDING:
-				return False
-			self._state = _CANCELLED
-			self._schedule_callbacks()
-			return True
-
-		def _schedule_callbacks(self):
-			"""Internal: Ask the event loop to call all callbacks.
-
-			The callbacks are scheduled to be called as soon as possible. Also
-			clears the callback list.
-			"""
-			callbacks = self._callbacks[:]
-			if not callbacks:
-				return
-
-			self._callbacks[:] = []
-			for callback in callbacks:
-				self._loop.call_soon(callback, self)
-
-		def cancelled(self):
-			"""Return True if the future was cancelled."""
-			return self._state == _CANCELLED
-
-		def done(self):
-			"""Return True if the future is done.
-
-			Done means either that a result / exception are available, or that the
-			future was cancelled.
-			"""
-			return self._state != _PENDING
-
-		def result(self):
-			"""Return the result this future represents.
-
-			If the future has been cancelled, raises CancelledError.  If the
-			future's result isn't yet available, raises InvalidStateError.  If
-			the future is done and has an exception set, this exception is raised.
-			"""
-			if self._state == _CANCELLED:
-				raise CancelledError()
-			if self._state != _FINISHED:
-				raise InvalidStateError('Result is not ready.')
-			if self._exception is not None:
-				raise self._exception
-			return self._result
-
-		def exception(self):
-			"""Return the exception that was set on this future.
-
-			The exception (or None if no exception was set) is returned only if
-			the future is done.  If the future has been cancelled, raises
-			CancelledError.  If the future isn't done yet, raises
-			InvalidStateError.
-			"""
-			if self._state == _CANCELLED:
-				raise CancelledError
-			if self._state != _FINISHED:
-				raise InvalidStateError('Exception is not set.')
-			return self._exception
-
-		def add_done_callback(self, fn):
-			"""Add a callback to be run when the future becomes done.
-
-			The callback is called with a single argument - the future object. If
-			the future is already done when this is called, the callback is
-			scheduled with call_soon.
-			"""
-			if self._state != _PENDING:
-				self._loop.call_soon(fn, self)
-			else:
-				self._callbacks.append(fn)
-
-		def remove_done_callback(self, fn):
-			"""Remove all instances of a callback from the "call when done" list.
-
-			Returns the number of callbacks removed.
-			"""
-			filtered_callbacks = [f for f in self._callbacks if f != fn]
-			removed_count = len(self._callbacks) - len(filtered_callbacks)
-			if removed_count:
-				self._callbacks[:] = filtered_callbacks
-			return removed_count
-
-		def set_result(self, result):
-			"""Mark the future done and set its result.
-
-			If the future is already done when this method is called, raises
-			InvalidStateError.
-			"""
-			if self._state != _PENDING:
-				raise InvalidStateError('{}: {!r}'.format(self._state, self))
-			self._result = result
-			self._state = _FINISHED
-			self._schedule_callbacks()
-
-		def set_exception(self, exception):
-			"""Mark the future done and set an exception.
-
-			If the future is already done when this method is called, raises
-			InvalidStateError.
-			"""
-			if self._state != _PENDING:
-				raise InvalidStateError('{}: {!r}'.format(self._state, self))
-			if isinstance(exception, type):
-				exception = exception()
-			self._exception = exception
-			self._state = _FINISHED
-			self._schedule_callbacks()
+	def _schedule_callbacks(self):
+		"""Internal: Ask the event loop to call all callbacks.
+
+		The callbacks are scheduled to be called as soon as possible. Also
+		clears the callback list.
+		"""
+		callbacks = self._callbacks[:]
+		if not callbacks:
+			return
+
+		self._callbacks[:] = []
+		for callback in callbacks:
+			self._loop.call_soon(callback, self)
+
+	def cancelled(self):
+		"""Return True if the future was cancelled."""
+		return self._state == _CANCELLED
+
+	def done(self):
+		"""Return True if the future is done.
+
+		Done means either that a result / exception are available, or that the
+		future was cancelled.
+		"""
+		return self._state != _PENDING
+
+	def result(self):
+		"""Return the result this future represents.
+
+		If the future has been cancelled, raises CancelledError.  If the
+		future's result isn't yet available, raises InvalidStateError.  If
+		the future is done and has an exception set, this exception is raised.
+		"""
+		if self._state == _CANCELLED:
+			raise CancelledError()
+		if self._state != _FINISHED:
+			raise InvalidStateError('Result is not ready.')
+		if self._exception is not None:
+			raise self._exception
+		return self._result
+
+	def exception(self):
+		"""Return the exception that was set on this future.
+
+		The exception (or None if no exception was set) is returned only if
+		the future is done.  If the future has been cancelled, raises
+		CancelledError.  If the future isn't done yet, raises
+		InvalidStateError.
+		"""
+		if self._state == _CANCELLED:
+			raise CancelledError
+		if self._state != _FINISHED:
+			raise InvalidStateError('Exception is not set.')
+		return self._exception
+
+	def add_done_callback(self, fn):
+		"""Add a callback to be run when the future becomes done.
+
+		The callback is called with a single argument - the future object. If
+		the future is already done when this is called, the callback is
+		scheduled with call_soon.
+		"""
+		if self._state != _PENDING:
+			self._loop.call_soon(fn, self)
+		else:
+			self._callbacks.append(fn)
+
+	def remove_done_callback(self, fn):
+		"""Remove all instances of a callback from the "call when done" list.
+
+		Returns the number of callbacks removed.
+		"""
+		filtered_callbacks = [f for f in self._callbacks if f != fn]
+		removed_count = len(self._callbacks) - len(filtered_callbacks)
+		if removed_count:
+			self._callbacks[:] = filtered_callbacks
+		return removed_count
+
+	def set_result(self, result):
+		"""Mark the future done and set its result.
+
+		If the future is already done when this method is called, raises
+		InvalidStateError.
+		"""
+		if self._state != _PENDING:
+			raise InvalidStateError('{}: {!r}'.format(self._state, self))
+		self._result = result
+		self._state = _FINISHED
+		self._schedule_callbacks()
+
+	def set_exception(self, exception):
+		"""Mark the future done and set an exception.
+
+		If the future is already done when this method is called, raises
+		InvalidStateError.
+		"""
+		if self._state != _PENDING:
+			raise InvalidStateError('{}: {!r}'.format(self._state, self))
+		if isinstance(exception, type):
+			exception = exception()
+		self._exception = exception
+		self._state = _FINISHED
+		self._schedule_callbacks()
 
 
 if Future is None:

[gentoo-commits] proj/portage:master commit in: pym/portage/util/futures/

[Zac Medico]

commit:     71c59145e0c7b631ec3f41e0d711445786d16f8f
Author:     Zac Medico <zmedico <AT> gentoo <DOT> org>
AuthorDate: Thu Mar 15 02:45:31 2018 +0000
Commit:     Zac Medico <zmedico <AT> gentoo <DOT> org>
CommitDate: Thu Mar 15 02:49:02 2018 +0000
URL:        https://gitweb.gentoo.org/proj/portage.git/commit/?id=71c59145

portage.util.futures.wait: fix arguments for asyncio compat

The bare "*" is not supported in python2.7, and in python3
the bare "*" means that keyword arguments must be used for
the arguments that follow.

Fixes: e43f6c583ed9 ("Add iter_completed convenience function (bug 648790)")

 pym/portage/util/futures/iter_completed.py |  2 +-
 pym/portage/util/futures/wait.py           | 16 +++-------------
 2 files changed, 4 insertions(+), 14 deletions(-)

diff --git a/pym/portage/util/futures/iter_completed.py b/pym/portage/util/futures/iter_completed.py
index 1050b6fa7..ad6275b49 100644
--- a/pym/portage/util/futures/iter_completed.py
+++ b/pym/portage/util/futures/iter_completed.py
@@ -52,7 +52,7 @@ def iter_completed(futures, max_jobs=None, max_load=None, loop=None):
 		# task_generator is exhausted
 		while future_map:
 			done, pending = loop.run_until_complete(
-				wait(*list(future_map.values()), return_when=FIRST_COMPLETED))
+				wait(list(future_map.values()), return_when=FIRST_COMPLETED))
 			for future in done:
 				del future_map[id(future)]
 				yield future

diff --git a/pym/portage/util/futures/wait.py b/pym/portage/util/futures/wait.py
index 3f0bdbff5..bd85bb053 100644
--- a/pym/portage/util/futures/wait.py
+++ b/pym/portage/util/futures/wait.py
@@ -11,15 +11,13 @@ except ImportError:
 from portage.util._eventloop.global_event_loop import global_event_loop
 
 
-# Use **kwargs since python2.7 does not allow arguments with defaults
-# to follow *futures.
-def wait(*futures, **kwargs):
+def wait(futures, loop=None, timeout=None, return_when=ALL_COMPLETED):
 	"""
 	Use portage's internal EventLoop to emulate asyncio.wait:
 	https://docs.python.org/3/library/asyncio-task.html#asyncio.wait
 
-	@param future: future to wait for
-	@type future: asyncio.Future (or compatible)
+	@param futures: futures to wait for
+	@type futures: asyncio.Future (or compatible)
 	@param timeout: number of seconds to wait (wait indefinitely if
 		not specified)
 	@type timeout: int or float
@@ -32,14 +30,6 @@ def wait(*futures, **kwargs):
 	@return: tuple of (done, pending).
 	@rtype: asyncio.Future (or compatible)
 	"""
-	if not futures:
-		raise TypeError("wait() missing 1 required positional argument: 'future'")
-	loop = kwargs.pop('loop', None)
-	timeout = kwargs.pop('timeout', None)
-	return_when = kwargs.pop('return_when', ALL_COMPLETED)
-	if kwargs:
-		raise TypeError("wait() got an unexpected keyword argument '{}'".\
-			format(next(iter(kwargs))))
 	loop = loop or global_event_loop()
 	result_future = loop.create_future()
 	_Waiter(futures, timeout, return_when, result_future, loop)

[gentoo-commits] proj/portage:master commit in: pym/portage/util/futures/

[Zac Medico]

commit:     be61882996099322bb3a1e82e71f475b4141ad40
Author:     Zac Medico <zmedico <AT> gentoo <DOT> org>
AuthorDate: Tue Apr 24 23:28:08 2018 +0000
Commit:     Zac Medico <zmedico <AT> gentoo <DOT> org>
CommitDate: Fri Apr 27 21:33:00 2018 +0000
URL:        https://gitweb.gentoo.org/proj/portage.git/commit/?id=be618829

Add iter_gather function (bug 653946)

This is similar to asyncio.gather, but takes an iterator of
futures as input, and includes support for max_jobs and max_load
parameters. For bug 653946, this will be used to asynchronously
gather the results of the portdbapi.async_fetch_map calls that
are required to generate a Manifest, while using the max_jobs
parameter to limit the number of concurrent async_aux_get calls.

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

 pym/portage/util/futures/iter_completed.py | 73 ++++++++++++++++++++++++++++++
 1 file changed, 73 insertions(+)

diff --git a/pym/portage/util/futures/iter_completed.py b/pym/portage/util/futures/iter_completed.py
index 5ad075305..1d6a9a4bd 100644
--- a/pym/portage/util/futures/iter_completed.py
+++ b/pym/portage/util/futures/iter_completed.py
@@ -112,3 +112,76 @@ def async_iter_completed(futures, max_jobs=None, max_load=None, loop=None):
 		# cleanup in case of interruption by SIGINT, etc
 		scheduler.cancel()
 		scheduler.wait()
+
+
+def iter_gather(futures, max_jobs=None, max_load=None, loop=None):
+	"""
+	This is similar to asyncio.gather, but takes an iterator of
+	futures as input, and includes support for max_jobs and max_load
+	parameters.
+
+	@param futures: iterator of asyncio.Future (or compatible)
+	@type futures: iterator
+	@param max_jobs: max number of futures to process concurrently (default
+		is multiprocessing.cpu_count())
+	@type max_jobs: int
+	@param max_load: max load allowed when scheduling a new future,
+		otherwise schedule no more than 1 future at a time (default
+		is multiprocessing.cpu_count())
+	@type max_load: int or float
+	@param loop: event loop
+	@type loop: EventLoop
+	@return: a Future resulting in a list of done input futures, in the
+		same order that they were yielded from the input iterator
+	@rtype: asyncio.Future (or compatible)
+	"""
+	loop = loop or global_event_loop()
+	loop = getattr(loop, '_asyncio_wrapper', loop)
+	result = loop.create_future()
+	futures_list = []
+
+	def future_generator():
+		for future in futures:
+			futures_list.append(future)
+			yield future
+
+	completed_iter = async_iter_completed(
+		future_generator(),
+		max_jobs=max_jobs,
+		max_load=max_load,
+		loop=loop,
+	)
+
+	def handle_result(future_done_set):
+		if result.cancelled():
+			if not future_done_set.cancelled():
+				# All exceptions must be consumed from future_done_set, in order
+				# to avoid triggering the event loop's exception handler.
+				list(future.exception() for future in future_done_set.result()
+					if not future.cancelled())
+			return
+
+		try:
+			handle_result.current_task = next(completed_iter)
+		except StopIteration:
+			result.set_result(futures_list)
+		else:
+			handle_result.current_task.add_done_callback(handle_result)
+
+	try:
+		handle_result.current_task = next(completed_iter)
+	except StopIteration:
+		handle_result.current_task = None
+		result.set_result(futures_list)
+	else:
+		handle_result.current_task.add_done_callback(handle_result)
+
+	def cancel_callback(result):
+		if (result.cancelled() and
+			handle_result.current_task is not None and
+			not handle_result.current_task.done()):
+			handle_result.current_task.cancel()
+
+	result.add_done_callback(cancel_callback)
+
+	return result