[gentoo-portage-dev] [PATCH 3/4] Add retry decorator (API inspired by tenacity)

[Zac Medico]

This decorator will be useful for retrying asynchronous
operations, such as gpg key refresh (bug 649276). The
API is inspired by tenacity, but is simpler. Only
asynchronous functions (like @asyncio.coroutine functions)
are supported. In order to retry a synchronous function,
first convert it to an asynchronous function as follows:

    asynchronous_func = functools.partial(
        loop.run_in_executor, None, synchronous_func)

Bug: https://bugs.gentoo.org/649276
See: https://github.com/jd/tenacity
---
 pym/portage/tests/util/futures/test_retry.py | 147 ++++++++++++++++++++++
 pym/portage/util/futures/futures.py          |   6 +
 pym/portage/util/futures/retry.py            | 178 +++++++++++++++++++++++++++
 3 files changed, 331 insertions(+)
 create mode 100644 pym/portage/tests/util/futures/test_retry.py
 create mode 100644 pym/portage/util/futures/retry.py

diff --git a/pym/portage/tests/util/futures/test_retry.py b/pym/portage/tests/util/futures/test_retry.py
new file mode 100644
index 000000000..7641e4e92
--- /dev/null
+++ b/pym/portage/tests/util/futures/test_retry.py
@@ -0,0 +1,147 @@
+# Copyright 2018 Gentoo Foundation
+# Distributed under the terms of the GNU General Public License v2
+
+import functools
+
+try:
+	import threading
+except ImportError:
+	import dummy_threading as threading
+
+from portage.tests import TestCase
+from portage.util._eventloop.global_event_loop import global_event_loop
+from portage.util.backoff import RandomExponentialBackoff
+from portage.util.futures.futures import TimeoutError
+from portage.util.futures.retry import retry
+from portage.util.futures.wait import wait
+from portage.util.monotonic import monotonic
+
+
+class SucceedLaterException(Exception):
+	pass
+
+
+class SucceedLater(object):
+	"""
+	A callable object that succeeds some duration of time has passed.
+	"""
+	def __init__(self, duration):
+		self._succeed_time = monotonic() + duration
+
+	def __call__(self):
+		remaining = self._succeed_time - monotonic()
+		if remaining > 0:
+			raise SucceedLaterException('time until success: {} seconds'.format(remaining))
+		return 'success'
+
+
+class SucceedNeverException(Exception):
+	pass
+
+
+class SucceedNever(object):
+	"""
+	A callable object that never succeeds.
+	"""
+	def __call__(self):
+		raise SucceedNeverException('expected failure')
+
+
+class HangForever(object):
+	"""
+	A callable object that sleeps forever.
+	"""
+	def __call__(self):
+		threading.Event().wait()
+
+
+class RetryTestCase(TestCase):
+	def testSucceedLater(self):
+		loop = global_event_loop()
+		func = SucceedLater(1)
+		func_coroutine = functools.partial(loop.run_in_executor, None, func)
+		decorator = retry(try_max=9999,
+			delay_func=RandomExponentialBackoff(multiplier=0.1, base=2))
+		decorated_func = decorator(func_coroutine)
+		result = loop.run_until_complete(decorated_func())
+		self.assertEqual(result, 'success')
+
+	def testSucceedNever(self):
+		loop = global_event_loop()
+		func = SucceedNever()
+		func_coroutine = functools.partial(loop.run_in_executor, None, func)
+		decorator = retry(try_max=4, try_timeout=None,
+			delay_func=RandomExponentialBackoff(multiplier=0.1, base=2))
+		decorated_func = decorator(func_coroutine)
+		done, pending = loop.run_until_complete(wait([decorated_func()]))
+		self.assertEqual(len(done), 1)
+		self.assertTrue(isinstance(done[0].exception().__cause__, SucceedNeverException))
+
+	def testSucceedNeverReraise(self):
+		loop = global_event_loop()
+		func = SucceedNever()
+		func_coroutine = functools.partial(loop.run_in_executor, None, func)
+		decorator = retry(reraise=True, try_max=4, try_timeout=None,
+			delay_func=RandomExponentialBackoff(multiplier=0.1, base=2))
+		decorated_func = decorator(func_coroutine)
+		done, pending = loop.run_until_complete(wait([decorated_func()]))
+		self.assertEqual(len(done), 1)
+		self.assertTrue(isinstance(done[0].exception(), SucceedNeverException))
+
+	def testHangForever(self):
+		loop = global_event_loop()
+		func = HangForever()
+		func_coroutine = functools.partial(loop.run_in_executor, None, func)
+		decorator = retry(try_max=2, try_timeout=0.1,
+			delay_func=RandomExponentialBackoff(multiplier=0.1, base=2))
+		decorated_func = decorator(func_coroutine)
+		done, pending = loop.run_until_complete(wait([decorated_func()]))
+		self.assertEqual(len(done), 1)
+		self.assertTrue(isinstance(done[0].exception().__cause__, TimeoutError))
+
+	def testHangForeverReraise(self):
+		loop = global_event_loop()
+		func = HangForever()
+		func_coroutine = functools.partial(loop.run_in_executor, None, func)
+		decorator = retry(reraise=True, try_max=2, try_timeout=0.1,
+			delay_func=RandomExponentialBackoff(multiplier=0.1, base=2))
+		decorated_func = decorator(func_coroutine)
+		done, pending = loop.run_until_complete(wait([decorated_func()]))
+		self.assertEqual(len(done), 1)
+		self.assertTrue(isinstance(done[0].exception(), TimeoutError))
+
+	def testCancelRetry(self):
+		loop = global_event_loop()
+		func = SucceedNever()
+		func_coroutine = functools.partial(loop.run_in_executor, None, func)
+		decorator = retry(try_timeout=0.1,
+			delay_func=RandomExponentialBackoff(multiplier=0.1, base=2))
+		decorated_func = decorator(func_coroutine)
+		future = decorated_func()
+		loop.call_later(0.3, future.cancel)
+		done, pending = loop.run_until_complete(wait([future]))
+		self.assertEqual(len(done), 1)
+		self.assertTrue(done[0].cancelled())
+
+	def testOverallTimeoutWithException(self):
+		loop = global_event_loop()
+		func = SucceedNever()
+		func_coroutine = functools.partial(loop.run_in_executor, None, func)
+		decorator = retry(try_timeout=0.1, overall_timeout=0.3,
+			delay_func=RandomExponentialBackoff(multiplier=0.1, base=2))
+		decorated_func = decorator(func_coroutine)
+		done, pending = loop.run_until_complete(wait([decorated_func()]))
+		self.assertEqual(len(done), 1)
+		self.assertTrue(isinstance(done[0].exception().__cause__, SucceedNeverException))
+
+	def testOverallTimeoutWithTimeoutError(self):
+		loop = global_event_loop()
+		# results in TimeoutError because it hangs forever
+		func = HangForever()
+		func_coroutine = functools.partial(loop.run_in_executor, None, func)
+		decorator = retry(try_timeout=0.1, overall_timeout=0.3,
+			delay_func=RandomExponentialBackoff(multiplier=0.1, base=2))
+		decorated_func = decorator(func_coroutine)
+		done, pending = loop.run_until_complete(wait([decorated_func()]))
+		self.assertEqual(len(done), 1)
+		self.assertTrue(isinstance(done[0].exception().__cause__, TimeoutError))
diff --git a/pym/portage/util/futures/futures.py b/pym/portage/util/futures/futures.py
index dcf593c01..cd56a27eb 100644
--- a/pym/portage/util/futures/futures.py
+++ b/pym/portage/util/futures/futures.py
@@ -11,6 +11,7 @@ __all__ = (
 	'CancelledError',
 	'Future',
 	'InvalidStateError',
+	'TimeoutError',
 )
 
 try:
@@ -18,6 +19,7 @@ try:
 		CancelledError,
 		Future,
 		InvalidStateError,
+		TimeoutError,
 	)
 except ImportError:
 
@@ -30,6 +32,10 @@ except ImportError:
 		def __init__(self):
 			Error.__init__(self, "cancelled")
 
+	class TimeoutError(Error):
+		def __init__(self):
+			Error.__init__(self, "timed out")
+
 	class InvalidStateError(Error):
 		pass
 
diff --git a/pym/portage/util/futures/retry.py b/pym/portage/util/futures/retry.py
new file mode 100644
index 000000000..2caf1bbac
--- /dev/null
+++ b/pym/portage/util/futures/retry.py
@@ -0,0 +1,178 @@
+# Copyright 2018 Gentoo Foundation
+# Distributed under the terms of the GNU General Public License v2
+
+import functools
+
+from portage.exception import PortageException
+from portage.util._eventloop.global_event_loop import global_event_loop
+from portage.util.futures.futures import TimeoutError
+
+
+class RetryError(PortageException):
+	"""Raised when retry fails."""
+	def __init__(self):
+		PortageException.__init__(self, "retry error")
+
+
+def retry(try_max=None, try_timeout=None, overall_timeout=None,
+	delay_func=None, reraise=False, loop=None):
+	"""
+	Create and return a retry decorator. The decorator is intended to
+	operate only on a coroutine function.
+
+	@param try_max: maximum number of tries
+	@type try_max: int or None
+	@param try_timeout: number of seconds to wait for a try to succeed
+		before cancelling it, which is only effective if func returns
+		tasks that support cancellation
+	@type try_timeout: float or None
+	@param overall_timeout: number of seconds to wait for retires to
+		succeed before aborting, which is only effective if func returns
+		tasks that support cancellation
+	@type overall_timeout: float or None
+	@param delay_func: function that takes an int argument corresponding
+		to the number of previous tries and returns a number of seconds
+		to wait before the next try
+	@type delay_func: callable
+	@param reraise: Reraise the last exception, instead of RetryError
+	@type reraise: bool
+	@param loop: event loop
+	@type loop: EventLoop
+	@return: func decorated with retry support
+	@rtype: callable
+	"""
+	return functools.partial(_retry_wrapper, loop, try_max, try_timeout,
+		overall_timeout, delay_func, reraise)
+
+
+def _retry_wrapper(loop, try_max, try_timeout, overall_timeout, delay_func,
+	reraise, func):
+	"""
+	Create and return a decorated function.
+	"""
+	return functools.partial(_retry, loop, try_max, try_timeout,
+		overall_timeout, delay_func, reraise, func)
+
+
+def _retry(loop, try_max, try_timeout, overall_timeout, delay_func,
+	reraise, func, *args, **kwargs):
+	"""
+	Retry coroutine, used to implement retry decorator.
+
+	@return: func return value
+	@rtype: asyncio.Future (or compatible)
+	"""
+	loop = loop or global_event_loop()
+	future = loop.create_future()
+	_Retry(future, loop, try_max, try_timeout, overall_timeout, delay_func,
+		reraise, functools.partial(func, *args, **kwargs))
+	return future
+
+
+class _Retry(object):
+	def __init__(self, future, loop, try_max, try_timeout, overall_timeout,
+		delay_func, reraise, func):
+		self._future = future
+		self._loop = loop
+		self._try_max = try_max
+		self._try_timeout = try_timeout
+		self._delay_func = delay_func
+		self._reraise = reraise
+		self._func = func
+
+		self._try_timeout_handle = None
+		self._overall_timeout_handle = None
+		self._overall_timeout_expired = None
+		self._tries = 0
+		self._current_task = None
+		self._previous_result = None
+
+		future.add_done_callback(self._cancel_callback)
+		if overall_timeout is not None:
+			self._overall_timeout_handle = loop.call_later(
+				overall_timeout, self._overall_timeout_callback)
+		self._begin_try()
+
+	def _cancel_callback(self, future):
+		if future.cancelled() and self._current_task is not None:
+			self._current_task.cancel()
+
+	def _try_timeout_callback(self):
+		self._try_timeout_handle = None
+		self._current_task.cancel()
+
+	def _overall_timeout_callback(self):
+		self._overall_timeout_handle = None
+		self._overall_timeout_expired = True
+		self._current_task.cancel()
+		self._retry_error()
+
+	def _begin_try(self):
+		self._tries += 1
+		self._current_task = self._func()
+		self._current_task.add_done_callback(self._try_done)
+		if self._try_timeout is not None:
+			self._try_timeout_handle = self._loop.call_later(
+				self._try_timeout, self._try_timeout_callback)
+
+	def _try_done(self, future):
+		self._current_task = None
+
+		if self._try_timeout_handle is not None:
+			self._try_timeout_handle.cancel()
+			self._try_timeout_handle = None
+
+		if not future.cancelled():
+			# consume exception, so that the event loop
+			# exception handler does not report it
+			future.exception()
+
+		if self._overall_timeout_expired:
+			return
+
+		try:
+			if self._future.cancelled():
+				return
+
+			self._previous_result = future
+			if not (future.cancelled() or future.exception() is not None):
+				# success
+				self._future.set_result(future.result())
+				return
+		finally:
+			if self._future.done() and self._overall_timeout_handle is not None:
+				self._overall_timeout_handle.cancel()
+				self._overall_timeout_handle = None
+
+		if self._try_max is not None and self._tries >= self._try_max:
+			self._retry_error()
+			return
+
+		if self._delay_func is not None:
+			delay = self._delay_func(self._tries)
+			self._current_task = self._loop.call_later(delay, self._delay_done)
+			return
+
+		self._begin_try()
+
+	def _delay_done(self):
+		self._current_task = None
+
+		if self._future.cancelled() or self._overall_timeout_expired:
+			return
+
+		self._begin_try()
+
+	def _retry_error(self):
+		if self._previous_result is None or self._previous_result.cancelled():
+			cause = TimeoutError()
+		else:
+			cause = self._previous_result.exception()
+
+		if self._reraise:
+			e = cause
+		else:
+			e = RetryError()
+			e.__cause__ = cause
+
+		self._future.set_exception(e)
-- 
2.13.6