diff options
Diffstat (limited to 'silx/utils')
| -rw-r--r-- | silx/utils/retry.py | 264 | ||||
| -rwxr-xr-x | silx/utils/test/__init__.py | 2 | ||||
| -rw-r--r-- | silx/utils/test/test_retry.py | 179 | ||||
| -rwxr-xr-x | silx/utils/test/test_testutils.py | 9 | ||||
| -rwxr-xr-x | silx/utils/testutils.py | 15 |
5 files changed, 467 insertions, 2 deletions
diff --git a/silx/utils/retry.py b/silx/utils/retry.py new file mode 100644 index 0000000..adc43bc --- /dev/null +++ b/silx/utils/retry.py @@ -0,0 +1,264 @@ +# coding: utf-8 +# /*########################################################################## +# Copyright (C) 2016-2017 European Synchrotron Radiation Facility +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +# THE SOFTWARE. +# +# ############################################################################*/ +""" +This module provides utility methods for retrying methods until they +no longer fail. +""" + +__authors__ = ["W. de Nolf"] +__license__ = "MIT" +__date__ = "05/02/2020" + + +import time +from functools import wraps +from contextlib import contextmanager +import multiprocessing +from queue import Empty + + +RETRY_PERIOD = 0.01 + + +class RetryTimeoutError(TimeoutError): + pass + + +class RetryError(RuntimeError): + pass + + +def _default_retry_on_error(e): + """ + :param BaseException e: + :returns bool: + """ + return isinstance(e, RetryError) + + +@contextmanager +def _handle_exception(options): + try: + yield + except BaseException as e: + retry_on_error = options.get("retry_on_error") + if retry_on_error is not None and retry_on_error(e): + options["exception"] = e + else: + raise + + +def _retry_loop(retry_timeout=None, retry_period=None, retry_on_error=None): + """Iterator which is endless or ends with an RetryTimeoutError. + It yields a dictionary which can be used to influence the loop. + + :param num retry_timeout: + :param num retry_period: sleep before retry + :param callable or None retry_on_error: checks whether an exception is + eligible for retry + """ + has_timeout = retry_timeout is not None + options = {"exception": None, "retry_on_error": retry_on_error} + if has_timeout: + t0 = time.time() + while True: + yield options + if retry_period is not None: + time.sleep(retry_period) + if has_timeout and (time.time() - t0) > retry_timeout: + raise RetryTimeoutError from options.get("exception") + + +def retry( + retry_timeout=None, retry_period=None, retry_on_error=_default_retry_on_error +): + """Decorator for a method that needs to be executed until it not longer + fails or until `retry_on_error` returns False. + + The decorator arguments can be overriden by using them when calling the + decorated method. + + :param num retry_timeout: + :param num retry_period: sleep before retry + :param callable or None retry_on_error: checks whether an exception is + eligible for retry + """ + + if retry_period is None: + retry_period = RETRY_PERIOD + + def decorator(method): + @wraps(method) + def wrapper(*args, **kw): + _retry_timeout = kw.pop("retry_timeout", retry_timeout) + _retry_period = kw.pop("retry_period", retry_period) + _retry_on_error = kw.pop("retry_on_error", retry_on_error) + for options in _retry_loop( + retry_timeout=_retry_timeout, + retry_period=_retry_period, + retry_on_error=_retry_on_error, + ): + with _handle_exception(options): + return method(*args, **kw) + + return wrapper + + return decorator + + +def retry_contextmanager( + retry_timeout=None, retry_period=None, retry_on_error=_default_retry_on_error +): + """Decorator to make a context manager from a method that needs to be + entered until it no longer fails or until `retry_on_error` returns False. + + The decorator arguments can be overriden by using them when calling the + decorated method. + + :param num retry_timeout: + :param num retry_period: sleep before retry + :param callable or None retry_on_error: checks whether an exception is + eligible for retry + """ + + if retry_period is None: + retry_period = RETRY_PERIOD + + def decorator(method): + @wraps(method) + def wrapper(*args, **kw): + _retry_timeout = kw.pop("retry_timeout", retry_timeout) + _retry_period = kw.pop("retry_period", retry_period) + _retry_on_error = kw.pop("retry_on_error", retry_on_error) + for options in _retry_loop( + retry_timeout=_retry_timeout, + retry_period=_retry_period, + retry_on_error=_retry_on_error, + ): + with _handle_exception(options): + gen = method(*args, **kw) + result = next(gen) + options["retry_on_error"] = None + yield result + try: + next(gen) + except StopIteration: + return + else: + raise RuntimeError(str(method) + " should only yield once") + + return contextmanager(wrapper) + + return decorator + + +def _subprocess_main(queue, method, retry_on_error, *args, **kw): + try: + result = method(*args, **kw) + except BaseException as e: + if retry_on_error(e): + # As the traceback gets lost, make sure the top-level + # exception is RetryError + e = RetryError(str(e)) + queue.put(e) + else: + queue.put(result) + + +def retry_in_subprocess( + retry_timeout=None, retry_period=None, retry_on_error=_default_retry_on_error +): + """Same as `retry` but it also retries segmentation faults. + + As subprocesses are spawned, you cannot use this decorator with the "@" syntax + because the decorated method needs to be an attribute of a module: + + .. code-block:: python + + def _method(*args, **kw): + ... + + method = retry_in_subprocess()(_method) + + :param num retry_timeout: + :param num retry_period: sleep before retry + :param callable or None retry_on_error: checks whether an exception is + eligible for retry + """ + + if retry_period is None: + retry_period = RETRY_PERIOD + + def decorator(method): + @wraps(method) + def wrapper(*args, **kw): + _retry_timeout = kw.pop("retry_timeout", retry_timeout) + _retry_period = kw.pop("retry_period", retry_period) + _retry_on_error = kw.pop("retry_on_error", retry_on_error) + + ctx = multiprocessing.get_context("spawn") + + def start_subprocess(): + queue = ctx.Queue(maxsize=1) + p = ctx.Process( + target=_subprocess_main, + args=(queue, method, retry_on_error) + args, + kwargs=kw, + ) + p.start() + return p, queue + + def stop_subprocess(p): + try: + p.kill() + except AttributeError: + p.terminate() + p.join() + + p, queue = start_subprocess() + try: + for options in _retry_loop( + retry_timeout=_retry_timeout, retry_on_error=_retry_on_error + ): + with _handle_exception(options): + if not p.is_alive(): + p, queue = start_subprocess() + try: + result = queue.get(block=True, timeout=_retry_period) + except Empty: + pass + except ValueError: + pass + else: + if isinstance(result, BaseException): + stop_subprocess(p) + raise result + else: + return result + finally: + stop_subprocess(p) + + return wrapper + + return decorator diff --git a/silx/utils/test/__init__.py b/silx/utils/test/__init__.py index 252bc05..b35feee 100755 --- a/silx/utils/test/__init__.py +++ b/silx/utils/test/__init__.py @@ -39,6 +39,7 @@ from . import test_number from . import test_external_resources from . import test_enum from . import test_testutils +from . import test_retry def suite(): @@ -54,4 +55,5 @@ def suite(): test_suite.addTest(test_external_resources.suite()) test_suite.addTest(test_enum.suite()) test_suite.addTest(test_testutils.suite()) + test_suite.addTest(test_retry.suite()) return test_suite diff --git a/silx/utils/test/test_retry.py b/silx/utils/test/test_retry.py new file mode 100644 index 0000000..d223f44 --- /dev/null +++ b/silx/utils/test/test_retry.py @@ -0,0 +1,179 @@ +# coding: utf-8 +# /*########################################################################## +# Copyright (C) 2016-2017 European Synchrotron Radiation Facility +# +# Permission is hereby granted, free of charge, to any person obtaining a copy +# of this software and associated documentation files (the "Software"), to deal +# in the Software without restriction, including without limitation the rights +# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +# copies of the Software, and to permit persons to whom the Software is +# furnished to do so, subject to the following conditions: +# +# The above copyright notice and this permission notice shall be included in +# all copies or substantial portions of the Software. +# +# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +# THE SOFTWARE. +# +# ############################################################################*/ +"""Tests for retry utilities""" + +__authors__ = ["W. de Nolf"] +__license__ = "MIT" +__date__ = "05/02/2020" + + +import unittest +import os +import sys +import time +import tempfile + +from .. import retry + + +def _cause_segfault(): + import ctypes + + i = ctypes.c_char(b"a") + j = ctypes.pointer(i) + c = 0 + while True: + j[c] = b"a" + c += 1 + + +def _submain(filename, kwcheck=None, ncausefailure=0, faildelay=0): + assert filename + assert kwcheck + sys.stderr = open(os.devnull, "w") + + with open(filename, mode="r") as f: + failcounter = int(f.readline().strip()) + + if failcounter < ncausefailure: + time.sleep(faildelay) + failcounter += 1 + with open(filename, mode="w") as f: + f.write(str(failcounter)) + if failcounter % 2: + raise retry.RetryError + else: + _cause_segfault() + return True + + +_wsubmain = retry.retry_in_subprocess()(_submain) + + +class TestRetry(unittest.TestCase): + def setUp(self): + self.test_dir = tempfile.mkdtemp() + self.ctr_file = os.path.join(self.test_dir, "failcounter.txt") + + def tearDown(self): + if os.path.exists(self.ctr_file): + os.unlink(self.ctr_file) + os.rmdir(self.test_dir) + + def test_retry(self): + ncausefailure = 3 + faildelay = 0.1 + sufficient_timeout = ncausefailure * (faildelay + 10) + insufficient_timeout = ncausefailure * faildelay * 0.5 + + @retry.retry() + def method(check, kwcheck=None): + assert check + assert kwcheck + nonlocal failcounter + if failcounter < ncausefailure: + time.sleep(faildelay) + failcounter += 1 + raise retry.RetryError + return True + + failcounter = 0 + kw = { + "kwcheck": True, + "retry_timeout": sufficient_timeout, + } + self.assertTrue(method(True, **kw)) + + failcounter = 0 + kw = { + "kwcheck": True, + "retry_timeout": insufficient_timeout, + } + with self.assertRaises(retry.RetryTimeoutError): + method(True, **kw) + + def test_retry_contextmanager(self): + ncausefailure = 3 + faildelay = 0.1 + sufficient_timeout = ncausefailure * (faildelay + 10) + insufficient_timeout = ncausefailure * faildelay * 0.5 + + @retry.retry_contextmanager() + def context(check, kwcheck=None): + assert check + assert kwcheck + nonlocal failcounter + if failcounter < ncausefailure: + time.sleep(faildelay) + failcounter += 1 + raise retry.RetryError + yield True + + failcounter = 0 + kw = {"kwcheck": True, "retry_timeout": sufficient_timeout} + with context(True, **kw) as result: + self.assertTrue(result) + + failcounter = 0 + kw = {"kwcheck": True, "retry_timeout": insufficient_timeout} + with self.assertRaises(retry.RetryTimeoutError): + with context(True, **kw) as result: + pass + + def test_retry_in_subprocess(self): + ncausefailure = 3 + faildelay = 0.1 + sufficient_timeout = ncausefailure * (faildelay + 10) + insufficient_timeout = ncausefailure * faildelay * 0.5 + + kw = { + "ncausefailure": ncausefailure, + "faildelay": faildelay, + "kwcheck": True, + "retry_timeout": sufficient_timeout, + } + with open(self.ctr_file, mode="w") as f: + f.write("0") + self.assertTrue(_wsubmain(self.ctr_file, **kw)) + + kw = { + "ncausefailure": ncausefailure, + "faildelay": faildelay, + "kwcheck": True, + "retry_timeout": insufficient_timeout, + } + with open(self.ctr_file, mode="w") as f: + f.write("0") + with self.assertRaises(retry.RetryTimeoutError): + _wsubmain(self.ctr_file, **kw) + + +def suite(): + test_suite = unittest.TestSuite() + test_suite.addTest(unittest.defaultTestLoader.loadTestsFromTestCase(TestRetry)) + return test_suite + + +if __name__ == "__main__": + unittest.main(defaultTest="suite") diff --git a/silx/utils/test/test_testutils.py b/silx/utils/test/test_testutils.py index c29a703..c72a3d8 100755 --- a/silx/utils/test/test_testutils.py +++ b/silx/utils/test/test_testutils.py @@ -84,6 +84,15 @@ class TestTestLogging(unittest.TestCase): logger.error("aaa") self.assertIsNotNone(listener) + def testErrorMessage(self): + logger = logging.getLogger(__name__ + "testCanBreak") + listener = testutils.TestLogging(logger, error=1, warning=2) + with self.assertRaisesRegex(RuntimeError, "aaabbb"): + with listener: + logger.error("aaa") + logger.warning("aaabbb") + logger.error("aaa") + def suite(): loadTests = unittest.defaultTestLoader.loadTestsFromTestCase diff --git a/silx/utils/testutils.py b/silx/utils/testutils.py index 1252269..434beee 100755 --- a/silx/utils/testutils.py +++ b/silx/utils/testutils.py @@ -102,6 +102,17 @@ def parameterize(test_case_class, *args, **kwargs): return suite +class LoggingRuntimeError(RuntimeError): + """Raised when the `TestLogging` fails""" + + def __init__(self, msg, records): + super(LoggingRuntimeError, self).__init__(msg) + self.records = records + + def __str__(self): + return super(LoggingRuntimeError, self).__str__() + " -> " + str(self.records) + + class TestLogging(logging.Handler): """Context checking the number of logging messages from a specified Logger. @@ -220,8 +231,8 @@ class TestLogging(logging.Handler): expected_count = expected_count_by_level[level] message += "%d %s (got %d)" % (expected_count, logging.getLevelName(level), count) - raise RuntimeError( - 'Expected %s' % message) + raise LoggingRuntimeError( + 'Expected %s' % message, records=list(self.records)) def emit(self, record): """Override :meth:`logging.Handler.emit`""" |