| """These basic tests may be applied to most types of executors.""" | |
| from __future__ import print_function | |
| from functools import partial | |
| from random import randint | |
| import traceback | |
| from threading import RLock | |
| import sys | |
| from concurrent.futures import Executor, CancelledError, wait, FIRST_COMPLETED | |
| from six.moves.queue import Queue | |
| try: | |
| from time import monotonic | |
| except ImportError: | |
| from monotonic import monotonic | |
| from hamcrest import ( | |
| assert_that, | |
| equal_to, | |
| calling, | |
| raises, | |
| instance_of, | |
| has_length, | |
| is_, | |
| contains_string, | |
| ) | |
| from pytest import fixture, skip | |
| from more_executors import Executors, RetryPolicy, f_proxy, f_return | |
| from .util import assert_soon, run_or_timeout | |
| from .logging_util import dump_executor, add_debug_logging | |
| TIMEOUT = 20.0 | |
| class SimulatedError(RuntimeError): | |
| pass | |
| def map_noop(value): | |
| # Make MapExecutor pass everything through unchanged | |
| return value | |
| def flat_map_noop(executor, value): | |
| # Make FlatMapExecutor pass everything through unchanged, | |
| # using the given executor to produce new futures | |
| return executor.submit(map_noop, value) | |
| def poll_noop(ds): | |
| # Make PollExecutor pass everything through unchanged | |
| [d.yield_result(d.result) for d in ds] | |
| # Poll again soon | |
| return 0.0001 | |
| # Don't want to make this public API now, but do want to test if | |
| # an executor returning everything as a proxy would work | |
| class ProxyExecutor(Executor): | |
| # pylint: disable=arguments-differ | |
| def __init__(self, delegate): | |
| self.__delegate = delegate | |
| def submit(self, *args, **kwargs): | |
| return f_proxy(self.__delegate.submit(*args, **kwargs)) | |
| def shutdown(self, *args, **kwargs): | |
| return self.__delegate.shutdown(*args, **kwargs) | |
| @fixture | |
| def retry_executor_ctor(): | |
| return lambda: Executors.thread_pool(name="test-retry").with_retry(max_attempts=1) | |
| @fixture | |
| def threadpool_executor_ctor(): | |
| return lambda: Executors.thread_pool(max_workers=20) | |
| @fixture | |
| def sync_executor_ctor(): | |
| return Executors.sync | |
| @fixture | |
| def map_executor_ctor(threadpool_executor_ctor): | |
| return lambda: threadpool_executor_ctor().with_map(map_noop) | |
| @fixture | |
| def flat_map_executor_ctor(threadpool_executor_ctor): | |
| def out(): | |
| threadpool_executor = threadpool_executor_ctor() | |
| return threadpool_executor.with_flat_map( | |
| partial(flat_map_noop, threadpool_executor) | |
| ) | |
| return out | |
| @fixture | |
| def proxy_executor_ctor(threadpool_executor_ctor): | |
| def out(): | |
| threadpool_executor = threadpool_executor_ctor() | |
| return ProxyExecutor(threadpool_executor) | |
| return out | |
| @fixture | |
| def throttle_executor_ctor(threadpool_executor_ctor): | |
| return lambda: threadpool_executor_ctor().with_throttle(10) | |
| @fixture | |
| def cancel_on_shutdown_executor_ctor(threadpool_executor_ctor): | |
| return lambda: threadpool_executor_ctor().with_cancel_on_shutdown() | |
| @fixture | |
| def map_retry_executor_ctor(threadpool_executor_ctor): | |
| return ( | |
| lambda: threadpool_executor_ctor().with_retry(RetryPolicy()).with_map(map_noop) | |
| ) | |
| @fixture | |
| def retry_map_executor_ctor(threadpool_executor_ctor): | |
| return ( | |
| lambda: threadpool_executor_ctor().with_map(map_noop).with_retry(RetryPolicy()) | |
| ) | |
| @fixture | |
| def timeout_executor_ctor(threadpool_executor_ctor): | |
| return lambda: threadpool_executor_ctor().with_timeout(60.0) | |
| @fixture | |
| def cancel_poll_map_retry_executor_ctor(threadpool_executor_ctor): | |
| return ( | |
| lambda: threadpool_executor_ctor() | |
| .with_retry(RetryPolicy()) | |
| .with_map(map_noop) | |
| .with_poll(poll_noop) | |
| .with_cancel_on_shutdown() | |
| ) | |
| @fixture | |
| def cancel_retry_map_poll_executor_ctor(threadpool_executor_ctor): | |
| return ( | |
| lambda: threadpool_executor_ctor() | |
| .with_poll(poll_noop) | |
| .with_map(map_noop) | |
| .with_retry(RetryPolicy()) | |
| .with_cancel_on_shutdown() | |
| ) | |
| @fixture | |
| def retry_map_poll_executor_ctor(threadpool_executor_ctor): | |
| return ( | |
| lambda: threadpool_executor_ctor() | |
| .with_poll(poll_noop) | |
| .with_map(map_noop) | |
| .with_retry(RetryPolicy()) | |
| ) | |
| def random_cancel(_value): | |
| """cancel function for use with poll executor which randomly decides whether | |
| cancel should succeed. This targets the stress test. The point here is that | |
| the futures should still satisfy the invariants of the Future API regardless | |
| of what the cancel function does.""" | |
| select = randint(0, 300) | |
| if select < 100: | |
| return True | |
| if select < 200: | |
| return False | |
| raise RuntimeError("simulated error from cancel") | |
| @fixture | |
| def poll_executor_ctor(threadpool_executor_ctor): | |
| return lambda: threadpool_executor_ctor().with_poll(poll_noop, random_cancel) | |
| def everything_executor(base_executor, name): | |
| # Get ready to go *nuts* | |
| return ( | |
| base_executor.with_poll(poll_noop, name=name) | |
| .with_map(map_noop) | |
| .with_retry(RetryPolicy()) | |
| .with_cancel_on_shutdown() | |
| .with_retry(max_attempts=1, max_sleep=0.1) | |
| .with_retry(RetryPolicy()) | |
| .with_throttle(10) | |
| .with_flat_map(partial(flat_map_noop, base_executor)) | |
| .with_timeout(120.0) | |
| .with_poll(poll_noop) | |
| .with_poll(poll_noop) | |
| .with_cancel_on_shutdown() | |
| .with_flat_map(lambda x: f_proxy(f_return(x))) | |
| .with_throttle(16) | |
| .with_flat_map(partial(flat_map_noop, base_executor)) | |
| .with_map(map_noop) | |
| .with_timeout(180.0) | |
| .with_map(map_noop) | |
| .with_retry(RetryPolicy()) | |
| ) | |
| @fixture | |
| def everything_sync_executor_ctor(sync_executor_ctor): | |
| return lambda: everything_executor(sync_executor_ctor(), "everything-sync") | |
| @fixture | |
| def everything_threadpool_executor_ctor(threadpool_executor_ctor): | |
| return lambda: everything_executor( | |
| threadpool_executor_ctor(), "everything-threadpool" | |
| ) | |
| EXECUTOR_TYPES = [ | |
| "threadpool", | |
| "retry", | |
| "map", | |
| "retry_map", | |
| "map_retry", | |
| "poll", | |
| "retry_map_poll", | |
| "sync", | |
| "timeout", | |
| "throttle", | |
| "cancel_poll_map_retry", | |
| "cancel_retry_map_poll", | |
| "flat_map", | |
| "proxy", | |
| "everything_sync", | |
| "everything_threadpool", | |
| ] | |
| @fixture(params=EXECUTOR_TYPES) | |
| def any_executor(request): | |
| ctor = request.getfixturevalue(request.param + "_executor_ctor") | |
| ex = ctor() | |
| # Capture log messages onto the executor itself, | |
| # for use with dump_executor if test fails. | |
| add_debug_logging(ex) | |
| # Want to know if the test failed; is there a better way | |
| # than counting the failure counts here?? | |
| failed_before = request.session.testsfailed | |
| yield ex | |
| failed_diff = request.session.testsfailed - failed_before | |
| if not failed_diff: | |
| try: | |
| kwargs = {} | |
| if sys.version_info > (3, 9): | |
| kwargs["cancel_futures"] = True | |
| run_or_timeout(ex.shutdown, True, **kwargs) | |
| return | |
| except Exception: | |
| print("Shutdown failed") | |
| dump_executor(ex) | |
| raise | |
| # Current test failed: | |
| # - dump state of executor for improved debugging | |
| # - use non-blocking shutdown, as blocking has a high chance | |
| # of hanging | |
| print("Test failed at time %s" % monotonic()) | |
| dump_executor(ex) | |
| ex.shutdown(False) | |
| @fixture(params=EXECUTOR_TYPES) | |
| def any_executor_ctor(request): | |
| return request.getfixturevalue(request.param + "_executor_ctor") | |
| def test_submit_results(any_executor): | |
| values = range(0, 1000) | |
| expected_results = [v * 2 for v in values] | |
| def fn(x): | |
| return x * 2 | |
| futures = [any_executor.submit(fn, x) for x in values] | |
| for f in futures: | |
| assert_that(not f.cancelled()) | |
| results = [f.result(TIMEOUT) for f in futures] | |
| assert_that(results, equal_to(expected_results)) | |
| def test_future_outlive_executor(any_executor_ctor): | |
| def make_futures(executor): | |
| return [executor.submit(lambda x: x * 2, y) for y in [1, 2, 3, 4]] | |
| futures = make_futures(any_executor_ctor()) | |
| results = [f.result(TIMEOUT) for f in futures] | |
| assert results == [2, 4, 6, 8] | |
| def test_broken_callback(any_executor): | |
| values = range(0, 1000) | |
| expected_results = [v * 2 for v in values] | |
| callback_calls = [] | |
| def fn(x): | |
| return x * 2 | |
| def broken_callback(f): | |
| callback_calls.append(f) | |
| raise RuntimeError("simulated broken callback") | |
| futures = [any_executor.submit(fn, x) for x in values] | |
| for f in futures: | |
| try: | |
| f.add_done_callback(broken_callback) | |
| except RuntimeError: | |
| # This is allowed - if future is done already, | |
| # the callback was invoked directly without exception handler. | |
| pass | |
| for f in futures: | |
| assert_that(not f.cancelled()) | |
| results = [f.result(TIMEOUT) for f in futures] | |
| assert_that(results, equal_to(expected_results)) | |
| # assert_soon as there's no guarantee that callbacks | |
| # are invoked before result() returns. | |
| assert_soon(lambda: assert_that(callback_calls, has_length(len(futures)))) | |
| for f in futures: | |
| assert_that(f in callback_calls) | |
| def test_submit_delayed_results(any_executor, request): | |
| if "sync" in request.node.name: | |
| skip("test not applicable with sync executor") | |
| values = [1, 2, 3] | |
| expected_results = [2, 4, 6] | |
| queue = Queue() | |
| def fn(value): | |
| queue.get(True) | |
| return value * 2 | |
| futures = [any_executor.submit(fn, x) for x in values] | |
| for f in futures: | |
| assert_that(not f.cancelled()) | |
| # They're not guaranteed to be "running" yet, but should | |
| # become so soon | |
| assert_soon(lambda: assert_that(all([f.running() for f in futures]))) | |
| # OK, they're not done yet though. | |
| for f in futures: | |
| assert_that(not f.done()) | |
| # Let them proceed | |
| queue.put(None) | |
| queue.put(None) | |
| queue.put(None) | |
| results = [f.result(TIMEOUT) for f in futures] | |
| assert_that(results, equal_to(expected_results)) | |
| def test_cancel(any_executor): | |
| for _ in range(0, 100): | |
| values = [1, 2, 3] | |
| expected_results = set([2, 4, 6]) | |
| def fn(x): | |
| return x * 2 | |
| futures = [any_executor.submit(fn, x) for x in values] | |
| cancelled = [] | |
| for f in futures: | |
| # There's no way we can be sure if cancel is possible. | |
| # We can only try... | |
| if f.cancel(): | |
| cancelled.append(f) | |
| assert_that(f.cancelled(), str(f)) | |
| assert_that(not f.running(), str(f)) | |
| assert_that(f.done(), str(f)) | |
| # Cancelling multiple times should be fine | |
| assert_that(f.cancel()) | |
| assert_that(f.cancel()) | |
| else: | |
| assert_that(not f.cancelled(), str(f)) | |
| for f in futures: | |
| if f in cancelled: | |
| assert_that( | |
| calling(f.result).with_args(TIMEOUT), raises(CancelledError), str(f) | |
| ) | |
| else: | |
| result = f.result(TIMEOUT) | |
| assert_that(result in expected_results) | |
| expected_results.remove(result) | |
| def test_blocked_cancel(any_executor, request): | |
| if "sync" in request.node.name: | |
| skip("test not applicable with sync executor") | |
| to_fn = Queue(1) | |
| from_fn = Queue(1) | |
| def fn(): | |
| to_fn.get() | |
| from_fn.put(None) | |
| to_fn.get() | |
| return 123 | |
| future = any_executor.submit(fn) | |
| # Wait until fn is certainly running | |
| to_fn.put(None) | |
| from_fn.get() | |
| # Since the function is in progress, | |
| # it should NOT be possible to cancel it | |
| assert_that(not future.cancel(), str(future)) | |
| # assert_soon since, by blocking on from_fn.get(), we only guarantee | |
| # that the future is running from the innermost executor/future's point | |
| # of view, but this may not have propagated to the outermost future yet | |
| assert_soon(lambda: assert_that(future.running(), str(future))) | |
| # Re-check cancel after running() is true | |
| assert_that(not future.cancel(), str(future)) | |
| # Let fn proceed and the future should be able to complete | |
| to_fn.put(None) | |
| assert_that(future.result(TIMEOUT), equal_to(123)) | |
| def get_traceback(future): | |
| exception = future.exception() | |
| if "__traceback__" in dir(exception): | |
| return exception.__traceback__ | |
| return future.exception_info()[1] | |
| def test_submit_mixed(any_executor): | |
| values = [1, 2, 3, 4] | |
| def crash_on_even(x): | |
| if (x % 2) == 0: | |
| raise SimulatedError("Simulated error on %s" % x) | |
| return x * 2 | |
| futures = [any_executor.submit(crash_on_even, x) for x in values] | |
| for f in futures: | |
| assert_that(not f.cancelled()) | |
| # Success | |
| assert_that(futures[0].result(TIMEOUT), equal_to(2)) | |
| # Crash, via exception | |
| assert_that(futures[1].exception(TIMEOUT), instance_of(SimulatedError)) | |
| assert_that( | |
| "".join(traceback.format_tb(get_traceback(futures[1]))), | |
| contains_string("crash_on_even"), | |
| ) | |
| # Success | |
| assert_that(futures[2].result(TIMEOUT), equal_to(6)) | |
| # Crash, via result | |
| assert_that( | |
| calling(futures[3].result).with_args(TIMEOUT), | |
| raises(SimulatedError, "Simulated error on 4"), | |
| ) | |
| def test_submit_staggered(any_executor, request): | |
| if "sync" in request.node.name: | |
| skip("test not applicable with sync executor") | |
| for _ in range(0, 100): | |
| do_test_submit_staggered(any_executor) | |
| def test_double_shutdown(any_executor): | |
| """Shutting down an executore more than once is OK.""" | |
| any_executor.shutdown() | |
| any_executor.shutdown() | |
| def test_submit_after_shutdown(any_executor): | |
| """Submitting after executor shutdown raises an error.""" | |
| any_executor.shutdown() | |
| assert_that( | |
| calling(any_executor.submit).with_args(lambda: 123), | |
| raises(RuntimeError, "cannot schedule new futures after shutdown"), | |
| ) | |
| def do_test_submit_staggered(executor): | |
| values = [1, 2, 3] | |
| expected_results = [2, 4, 6, 2, 4, 6] | |
| q1 = Queue() | |
| q2 = Queue() | |
| def fn(value): | |
| q1.get(True) | |
| q2.get(True) | |
| return value * 2 | |
| futures = [executor.submit(fn, x) for x in values] | |
| for f in futures: | |
| assert_that(not f.cancelled()) | |
| # They're not guaranteed to be "running" yet, but should | |
| # become so soon | |
| assert_soon(lambda: assert_that(all([f.running() for f in futures]))) | |
| # OK, they're not done yet though. | |
| for f in futures: | |
| assert_that(not f.done()) | |
| # Let them proceed to first checkpoint | |
| [q1.put(None) for f in futures] | |
| # Submit some more | |
| futures.extend([executor.submit(fn, x) for x in values]) | |
| # Let a couple of futures complete | |
| q2.put(True) | |
| q2.put(True) | |
| done, not_done = wait(futures, return_when=FIRST_COMPLETED, timeout=TIMEOUT) | |
| # Might have received 1, or 2 | |
| if len(done) == 1: | |
| more_done, _more_not_done = wait( | |
| not_done, return_when=FIRST_COMPLETED, timeout=TIMEOUT | |
| ) | |
| done = done | more_done | |
| assert_that(done, has_length(2)) | |
| for f in done: | |
| assert_that(f.done(), str(f)) | |
| # OK, let them all finish up now | |
| [q1.put(None) for _ in (1, 2, 3)] | |
| [q2.put(None) for _ in (1, 2, 3, 4)] | |
| results = [f.result(TIMEOUT) for f in futures] | |
| assert_that(results, equal_to(expected_results)) | |
| class StressTester(object): | |
| FUTURES_LIMIT = 500 | |
| CANCELLED = object() | |
| def __init__(self, executor): | |
| self.executor = executor | |
| self.lock = RLock() | |
| self.futures = [] | |
| self.future_idents = {} | |
| self.expected_results = {} | |
| self.idents = 0 | |
| def next_ident(self, msg): | |
| with self.lock: | |
| self.idents = self.idents + 1 | |
| return "%s %d" % (msg, self.idents) | |
| def cancel_something(self): | |
| # Try to pick and cancel some future | |
| for f in self.futures: | |
| if f.cancel(): | |
| with self.lock: | |
| ident = self.future_idents[f] | |
| self.expected_results[ident] = self.CANCELLED | |
| return | |
| def stress_fn(self, ident, behavior): | |
| sub_future = None | |
| if len(self.futures) < self.FUTURES_LIMIT: | |
| sub_ident = self.next_ident("submit from [%s]" % ident) | |
| sub_future = self.executor.submit(self.stress_fn, sub_ident, randint(0, 3)) | |
| self.add_future(sub_future, sub_ident) | |
| # Return a value | |
| if behavior == 0: | |
| with self.lock: | |
| assert ident not in self.expected_results | |
| self.expected_results[ident] = ident | |
| return ident | |
| # Raise an exception | |
| if behavior == 1: | |
| error = RuntimeError("error %s" % ident) | |
| with self.lock: | |
| assert ident not in self.expected_results | |
| self.expected_results[ident] = error | |
| raise error | |
| # Submit again from callback | |
| if behavior == 2 and sub_future: | |
| sub_future.add_done_callback( | |
| lambda f: self.stress_fn( | |
| self.next_ident("case2 from [%s]" % sub_ident), 2 | |
| ) | |
| ) | |
| with self.lock: | |
| assert ident not in self.expected_results | |
| self.expected_results[ident] = ident * 3 | |
| return ident * 3 | |
| if behavior == 3: | |
| self.cancel_something() | |
| with self.lock: | |
| self.expected_results[ident] = ident * 4 | |
| return ident * 4 | |
| self.expected_results[ident] = ident * 5 | |
| return ident * 5 | |
| def add_future(self, future, ident): | |
| with self.lock: | |
| if len(self.futures) < self.FUTURES_LIMIT: | |
| self.futures.append(future) | |
| self.future_idents[future] = ident | |
| return True | |
| future.cancel() | |
| return False | |
| def start(self): | |
| for _ in range(0, 100): | |
| value = randint(0, 3) | |
| ident = self.next_ident("init") | |
| future = self.executor.submit(self.stress_fn, ident, value) | |
| if not self.add_future(future, ident): | |
| break | |
| def verify(self): | |
| # Wait until all the expected futures have been created | |
| assert_soon( | |
| lambda: assert_that(len(self.futures), equal_to(self.FUTURES_LIMIT)) | |
| ) | |
| # The timeout here is so that the test fails rather than hangs forever, | |
| # if something goes wrong. | |
| done, _not_done = wait(self.futures, 60.0) | |
| assert_that(len(done), equal_to(len(self.futures))) | |
| for f in self.futures: | |
| self.verify_future(f) | |
| def verify_future(self, f): | |
| assert f.done(), str(f) | |
| ident = self.future_idents[f] | |
| assert ident in self.expected_results, "missing entry %s for future %s" % ( | |
| ident, | |
| f, | |
| ) | |
| expected_result = self.expected_results[ident] | |
| if expected_result is self.CANCELLED: | |
| assert_that(f.cancelled()) | |
| elif isinstance(expected_result, Exception): | |
| assert_that(f.exception(TIMEOUT), is_(expected_result)) | |
| else: | |
| assert_that(f.result(TIMEOUT), equal_to(expected_result)) | |
| def test_stress(any_executor, request): | |
| if "sync" in request.node.name: | |
| # The test as written currently will blow the stack on sync executor | |
| skip("test not applicable with sync executor") | |
| tester = StressTester(any_executor) | |
| tester.start() | |
| tester.verify() |