X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=smart_future.py;h=a1f6d97741db715871d404d5a126d87a5601b503;hb=f2600f30801c849fc1d139386e3ddc3c9eb43e30;hp=604c149520464bcd9d8c5a55cf8905acd5ec34d4;hpb=f2b4fe83f6fc853a68653bd5e3d9fe0648c3d105;p=python_utils.git diff --git a/smart_future.py b/smart_future.py index 604c149..a1f6d97 100644 --- a/smart_future.py +++ b/smart_future.py @@ -1,18 +1,23 @@ #!/usr/bin/env python3 +"""A future that can be treated like the result that it contains and +will not block until it is used. At that point, if the underlying +value is not yet available, it will block until it becomes +available.""" + from __future__ import annotations import concurrent import concurrent.futures as fut import logging -import traceback -from typing import Callable, List, TypeVar +from typing import Callable, List, Set, TypeVar from overrides import overrides +import id_generator + # This module is commonly used by others in here and should avoid # taking any unnecessary dependencies back on them. from deferred_operand import DeferredOperand -import id_generator logger = logging.getLogger(__name__) @@ -27,11 +32,11 @@ def wait_any( ): real_futures = [] smart_future_by_real_future = {} - completed_futures = set() - for f in futures: - assert type(f) == SmartFuture - real_futures.append(f.wrapped_future) - smart_future_by_real_future[f.wrapped_future] = f + completed_futures: Set[fut.Future] = set() + for x in futures: + assert isinstance(x, SmartFuture) + real_futures.append(x.wrapped_future) + smart_future_by_real_future[x.wrapped_future] = x while len(completed_futures) != len(real_futures): newly_completed_futures = concurrent.futures.as_completed(real_futures) @@ -42,15 +47,12 @@ def wait_any( if log_exceptions and not f.cancelled(): exception = f.exception() if exception is not None: - logger.warning( - f'Future {id(f)} raised an unhandled exception and exited.' - ) + logger.warning('Future 0x%x raised an unhandled exception and exited.', id(f)) logger.exception(exception) raise exception yield smart_future_by_real_future[f] if callback is not None: callback() - return def wait_all( @@ -59,9 +61,9 @@ def wait_all( log_exceptions: bool = True, ) -> None: real_futures = [] - for f in futures: - assert type(f) == SmartFuture - real_futures.append(f.wrapped_future) + for x in futures: + assert isinstance(x, SmartFuture) + real_futures.append(x.wrapped_future) (done, not_done) = concurrent.futures.wait( real_futures, timeout=None, return_when=concurrent.futures.ALL_COMPLETED @@ -71,9 +73,7 @@ def wait_all( if not f.cancelled(): exception = f.exception() if exception is not None: - logger.warning( - f'Future {id(f)} raised an unhandled exception and exited.' - ) + logger.warning('Future 0x%x raised an unhandled exception and exited.', id(f)) logger.exception(exception) raise exception assert len(done) == len(real_futures) @@ -89,7 +89,7 @@ class SmartFuture(DeferredOperand): """ def __init__(self, wrapped_future: fut.Future) -> None: - assert type(wrapped_future) == fut.Future + assert isinstance(wrapped_future, fut.Future) self.wrapped_future = wrapped_future self.id = id_generator.get("smart_future_id") @@ -102,5 +102,5 @@ class SmartFuture(DeferredOperand): # You shouldn't have to call this; instead, have a look at defining a # method on DeferredOperand base class. @overrides - def _resolve(self, *, timeout=None) -> T: + def _resolve(self, timeout=None) -> T: return self.wrapped_future.result(timeout)