Merge pull request #72 from digital-asset/python-startup-fixes

python: Fix a bug that caused ready() calls to perpetually hang under certain circumstances

Author: da-tanabe

VERSION

@@ -1 +1 @@
-6.6.4
+6.7.0

python/dazl/client/_network_client_impl.py

@@ -20,18 +20,16 @@
 from ._base_model import IfMissingPartyBehavior, CREATE_IF_MISSING, NONE_IF_MISSING, \
     EXCEPTION_IF_MISSING
 from ._party_client_impl import _PartyClientImpl
-from ._run_level import RunState
 from .bots import Bot, BotCollection
-from .config import AnonymousNetworkConfig, NetworkConfig, URLConfig, \
-    DEFAULT_CONNECT_TIMEOUT_SECONDS
+from .config import AnonymousNetworkConfig, NetworkConfig, URLConfig
 from ..metrics import MetricEvents
-from ..model.core import Party, RunLevel, DazlPartyMissingError
+from ..model.core import Party, DazlPartyMissingError
 from ..model.ledger import LedgerMetadata
 from ..model.network import connection_settings
 from ..model.reading import InitEvent, ReadyEvent, BaseEvent
 from ..protocols import LedgerNetwork
 from ..protocols.autodetect import AutodetectLedgerNetwork
-from ..util.asyncio_util import execute_in_loop, Invoker, safe_create_future
+from ..scheduler import Invoker, RunLevel
 from ..util.dar import get_dar_package_ids
 from ..util.prim_types import to_timedelta, TimeDeltaConvertible
 
@@ -41,16 +39,15 @@
 class _NetworkImpl:
 
     __slots__ = ('_lock', '_main_thread', 'invoker', '_party_clients', '_global_impls',
-                 '_party_impls', '_run_state', 'bots', '_config', '_pool', '_pool_init',
+                 '_party_impls', 'bots', '_config', '_pool', '_pool_init',
                  '_cached_metadata', '_metrics')
 
     def __init__(self, metrics: 'Optional[MetricEvents]' = None):
         self.invoker = Invoker()
         self._lock = RLock()
         self._main_thread = None  # type: Optional[Thread]
-        self._run_state = None  # type: Optional[RunState]
         self._pool = None  # type: Optional[LedgerNetwork]
-        self._pool_init = safe_create_future()
+        self._pool_init = self.invoker.create_future()
         if metrics is None:
             # create a default set of metrics
             try:
@@ -71,27 +68,6 @@ def __init__(self, metrics: 'Optional[MetricEvents]' = None):
         self._config = dict()
         self.bots = BotCollection(None)
 
-    @overload
-    def run_in_loop_threadsafe(self,
-                               cb: Callable[[], Union[None, Awaitable[None]]],
-                               timeout: float = 30) -> None: ...
-
-    @overload
-    def run_in_loop_threadsafe(self,
-                               cb: Callable[[], Union[Awaitable[T], T]],
-                               timeout: float = 30) -> T: ...
-
-    def run_in_loop_threadsafe(self, cb, timeout=30):
-        """
-        Schedule a callback to be run on the event loop. This can either be a normal function or a
-        coroutine.
-
-        :param cb: The callback to invoke.
-        :param timeout: The timeout, in seconds, to abort.
-        :return: The returned value from the function or coroutine.
-        """
-        return self.invoker.run_in_loop(cb, timeout=timeout)
-
     def set_config(self, *configs: 'Union[NetworkConfig, AnonymousNetworkConfig]', **kwargs):
         for config in configs:
             self._config.update({k: v for k, v in asdict(config).items() if v is not None})
@@ -111,25 +87,18 @@ def resolved_config(self) -> 'NetworkConfig':
     def resolved_anonymous_config(self) -> 'AnonymousNetworkConfig':
         return AnonymousNetworkConfig.parse_kwargs(**self._config)
 
-    async def aio_run(self, *coroutines, run_state: Optional[RunState] = None) -> None:
+    async def aio_run(self, *coroutines) -> None:
         """
         Coroutine where all network activity is scheduled from.
 
         :param coroutines:
             Optional additional coroutines to run. ``aio_run`` is only considered complete once all
             of the additional coroutines are also finished.
-        :param run_state:
-            :class:`RunState` that specifies when the loop is to be terminated. If ``None`` is
-            supplied, a default :class:`RunState` that runs until explicitly stopped is assigned.
         """
         from ..metrics.instrumenters import AioLoopPerfMonitor
         from ..protocols import LedgerConnectionOptions
 
-        if run_state is None:
-            run_state = RunState(RunLevel.RUN_FOREVER)
-
         with self._lock:
-            self._run_state = run_state
             self.invoker.set_context_as_current()
             config = self.resolved_config()
 
@@ -157,21 +126,20 @@ async def aio_run(self, *coroutines, run_state: Optional[RunState] = None) -> No
 
             options = LedgerConnectionOptions(connect_timeout=connect_timeout)
 
-            self._pool = pool = AutodetectLedgerNetwork(
-                options=options, loop=self.invoker.get_loop(), executor=self.invoker.get_executor(),
-                run_state=self._run_state)
+            self._pool = pool = AutodetectLedgerNetwork(self.invoker, options)
             self._pool_init.set_result(pool)
 
             try:
-                runner = _NetworkRunner(pool, run_state, self, coroutines)
+                runner = _NetworkRunner(pool, self, coroutines)
                 await runner.run()
             finally:
                 await self._pool.close()
                 if site is not None:
                     await site.stop()
+                await self.invoker.shutdown(0)
                 self._pool = None
 
-    def start(self, run_state: RunState, daemon: bool) -> None:
+    def start(self, daemon: bool) -> None:
         """
         Create a background thread, start an event loop, and run the application.
         """
@@ -184,7 +152,7 @@ def background_main():
                 loop = new_event_loop()
                 set_event_loop(loop)
 
-                loop.run_until_complete(self.aio_run(run_state=run_state))
+                loop.run_until_complete(self.aio_run())
             except:
                 LOG.exception('The main event loop died!')
 
@@ -207,11 +175,7 @@ def shutdown(self) -> None:
         """
         with self._lock:
             LOG.info('Shutting down...')
-            loop = self.invoker.get_loop()
-            run_state = self._run_state
-            if loop is None:
-                raise RuntimeError('shutdown() called on a non-stopped Network')
-        loop.call_soon_threadsafe(run_state.handle_sigint)
+            self.invoker.handle_sigint()
 
     def abort(self) -> None:
         """
@@ -221,11 +185,7 @@ def abort(self) -> None:
         """
         with self._lock:
             LOG.info('Aborting...')
-            loop = self.invoker.get_loop()
-            run_state = self._run_state
-            if loop is None:
-                raise RuntimeError('abort() called on a non-stopped Network')
-        loop.call_soon_threadsafe(run_state.handle_sigquit)
+            self.invoker.handle_sigquit()
 
     def join(self, timeout=None):
         """
@@ -241,10 +201,10 @@ def join(self, timeout=None):
 
     def global_impl(self, ctor: 'Callable[[_NetworkImpl], T]') -> T:
         with self._lock:
-            inst = self._global_impls.get(ctor)
+            inst = self._global_impls.get(ctor)  # type: ignore
             if inst is None:
                 inst = ctor(self)
-                self._global_impls[ctor] = inst
+                self._global_impls[ctor] = inst  # type: ignore
             return inst
 
     @overload
@@ -342,7 +302,7 @@ def simple_metadata(self, timeout: 'TimeDeltaConvertible') -> LedgerMetadata:
             return self._cached_metadata
 
         with self._lock:
-            return execute_in_loop(self.invoker.get_loop(), self._pool.ledger, timeout=timeout)
+            return self.invoker.run_in_loop(self._pool.ledger, timeout=timeout)
 
     async def aio_metadata(self) -> LedgerMetadata:
         """
@@ -417,9 +377,8 @@ async def connect_anonymous(self):
 
 
 class _NetworkRunner:
-    def __init__(self, pool: LedgerNetwork, run_state: RunState, network_impl: '_NetworkImpl', user_coroutines):
+    def __init__(self, pool: LedgerNetwork, network_impl: '_NetworkImpl', user_coroutines):
         self.pool = pool
-        self.run_state = run_state
         self.initialized_parties = set()  # type: Set[Party]
         self._network_impl = network_impl
         self._config = self._network_impl.resolved_config()
@@ -483,10 +442,10 @@ async def beat(self, party_impls: 'Collection[_PartyClientImpl]') -> Tuple[str,
         self._user_coroutines = [fut for fut in self._user_coroutines if not fut.done()]
 
         # If there are no more commands in flight, there is no more activity
-        if self.run_state.level >= RunLevel.TERMINATE_GRACEFULLY:
+        if self._network_impl.invoker.level >= RunLevel.TERMINATE_GRACEFULLY:
             LOG.info('network_run terminating on user request...')
             return offset, False
-        elif self.run_state.level >= RunLevel.RUN_UNTIL_IDLE:
+        elif self._network_impl.invoker.level >= RunLevel.RUN_UNTIL_IDLE:
             if not self._read_completions and not self._user_coroutines:
                 if all(pi.writer_idle() for pi in party_impls):
                     LOG.info('network_run: terminating because all writers are idle and all '

python/dazl/client/_party_client_impl.py

@@ -26,8 +26,7 @@
     InitEvent, ReadyEvent, ActiveContractSetEvent, PackagesAddedEvent
 from ..model.writing import CommandBuilder, CommandDefaults, CommandPayload, EventHandlerResponse
 from ..protocols import LedgerNetwork, LedgerClient
-from ..util.asyncio_util import ServiceQueue, await_then, completed, safe_create_future, \
-    named_gather
+from ..util.asyncio_util import ServiceQueue, await_then, completed, named_gather
 from ..util.prim_natural import n_things
 from ..util.typing import safe_cast
 
@@ -51,10 +50,10 @@ def __init__(self, parent: '_NetworkImpl', party: 'Party'):
         self._pool = None  # type: Optional[LedgerNetwork]
         self._pool_fut = None  # type: Optional[Awaitable[LedgerNetwork]]
         self._client_fut = None  # type: Optional[Awaitable[LedgerClient]]
-        self._ready_fut = safe_create_future()
+        self._ready_fut = self.invoker.create_future()
         self._known_packages = set()  # type: Set[str]
 
-        self._acs = ActiveContractSet()
+        self._acs = ActiveContractSet(self.invoker)
         self.bots = BotCollection(party)
         self._reader = _PartyClientReaderState()
         self._writer = _PartyClientWriterState()