smart_future.py

   1 #!/usr/bin/env python3
   2
   3 from __future__ import annotations
   4 from collections.abc import Mapping
   5 import concurrent.futures as fut
   6 import time
   7 from typing import Callable, List, TypeVar
   8
   9 from deferred_operand import DeferredOperand
  10 import id_generator
  11
  12 T = TypeVar('T')
  13
  14
  15 def wait_many(futures: List[SmartFuture], *, callback: Callable = None):
  16     finished: Mapping[int, bool] = {}
  17     x = 0
  18     while True:
  19         future = futures[x]
  20         if not finished.get(future.get_id(), False):
  21             if future.is_ready():
  22                 finished[future.get_id()] = True
  23                 yield future
  24             else:
  25                 if callback is not None:
  26                     callback()
  27                 time.sleep(0.1)
  28         x += 1
  29         if x >= len(futures):
  30             x = 0
  31         if len(finished) == len(futures):
  32             if callback is not None:
  33                 callback()
  34             return
  35
  36
  37 class SmartFuture(DeferredOperand):
  38     """This is a SmartFuture, a class that wraps a normal Future and can
  39     then be used, mostly, like a normal (non-Future) identifier.
  40
  41     Using a FutureWrapper in expressions will block and wait until
  42     the result of the deferred operation is known.
  43     """
  44
  45     def __init__(self, wrapped_future: fut.Future) -> None:
  46         self.wrapped_future = wrapped_future
  47         self.id = id_generator.get("smart_future_id")
  48
  49     def get_id(self) -> int:
  50         return self.id
  51
  52     def is_ready(self) -> bool:
  53         return self.wrapped_future.done()
  54
  55     # You shouldn't have to call this; instead, have a look at defining a
  56     # method on DeferredOperand base class.
  57     def _resolve(self, *, timeout=None) -> T:
  58         return self.wrapped_future.result(timeout)