Add BaseService.is_operational

is_operational: launch preamble complete and no cancel triggered

Use it everywhere that used to have service.is_running or
service.cancel_token.triggered

Author: carver

p2p/DEVELOPMENT.md

@@ -18,7 +18,7 @@ library.
 ## BaseService
 
 - If your service needs to run coroutines in the background, you should use the `BaseService.run_task()` method and
-  ensure they exit when `is_running` is False or when the cancel token is triggered.
+  ensure they exit when `is_operational` is False or when the cancel token is triggered.
 - If your service runs other services in the background, you should pass your CancelToken down to
   those services and run those using `BaseService.run_child_service()`, or
   `BaseService.run_daemon()` if you want the parent to be terminated when the child dies

p2p/discovery.py

@@ -337,7 +337,7 @@ async def _run(self) -> None:
         await self._start_udp_listener()
         connect_loop_sleep = 2
         self.run_task(self.proto.bootstrap())
-        while not self.cancel_token.triggered:
+        while self.is_operational:
             await self.maybe_connect_to_more_peers()
             await self.sleep(connect_loop_sleep)
 

p2p/nat.py

@@ -72,7 +72,7 @@ async def _run(self) -> None:
         On every iteration we configure the port mapping with a lifetime of 30 minutes and then
         sleep for that long as well.
         """
-        while not self.cancel_token.triggered:
+        while self.is_operational:
             try:
                 # Wait for the port mapping lifetime, and then try registering it again
                 await self.wait(asyncio.sleep(self._nat_portmap_lifetime))

p2p/peer.py

@@ -342,7 +342,7 @@ async def _cleanup(self) -> None:
         self.close()
 
     async def _run(self) -> None:
-        while not self.cancel_token.triggered:
+        while self.is_operational:
             try:
                 cmd, msg = await self.read_msg()
             except (PeerConnectionLost, TimeoutError) as err:
@@ -530,7 +530,7 @@ async def disconnect(self, reason: DisconnectReason) -> None:
         self.logger.debug("Disconnecting from remote peer; reason: %s", reason.name)
         self.base_protocol.send_disconnect(reason.value)
         self.close()
-        if self.is_running:
+        if self.is_operational:
             await self.cancel()
 
     def select_sub_protocol(self, remote_capabilities: List[Tuple[bytes, int]]
@@ -912,7 +912,7 @@ def get_peers(self, min_td: int) -> List[BasePeer]:
         return [peer for peer in peers if peer.head_td >= min_td]
 
     async def _periodically_report_stats(self) -> None:
-        while self.is_running:
+        while self.is_operational:
             inbound_peers = len(
                 [peer for peer in self.connected_nodes.values() if peer.inbound])
             self.logger.info("Connected peers: %d inbound, %d outbound",

p2p/service.py

@@ -89,7 +89,7 @@ async def run(
         """
         if self.is_running:
             raise RuntimeError("Cannot start the service while it's already running")
-        elif self.cancel_token.triggered:
+        elif self.is_cancelled:
             raise RuntimeError("Cannot restart a service that has already been cancelled")
 
         if finished_callback:
@@ -135,6 +135,7 @@ async def _run_task_wrapper() -> None:
                 pass
             except Exception as e:
                 self.logger.warning("Task %s finished unexpectedly: %s", awaitable, e)
+                self.logger.warning("Task failure traceback", exc_info=True)
             else:
                 self.logger.debug("Task %s finished with no errors", awaitable)
         self._tasks.add(asyncio.ensure_future(_run_task_wrapper()))
@@ -158,7 +159,7 @@ async def _run_daemon_wrapper() -> None:
             try:
                 await service.run()
             finally:
-                if not self.cancel_token.triggered:
+                if not self.is_cancelled:
                     self.logger.debug(
                         "%s finished while we're still running, terminating as well", service)
                     self.cancel_token.trigger()
@@ -188,7 +189,7 @@ async def cleanup(self) -> None:
 
     async def cancel(self) -> None:
         """Trigger the CancelToken and wait for the cleaned_up event to be set."""
-        if self.cancel_token.triggered:
+        if self.is_cancelled:
             self.logger.warning("Tried to cancel %s, but it was already cancelled", self)
             return
         elif not self.is_running:
@@ -215,6 +216,14 @@ def _forcibly_cancel_all_tasks(self) -> None:
         for task in self._tasks:
             task.cancel()
 
+    @property
+    def is_cancelled(self) -> bool:
+        return self.cancel_token.triggered
+
+    @property
+    def is_operational(self) -> bool:
+        return self.events.started.is_set() and not self.cancel_token.triggered
+
     @property
     def is_running(self) -> bool:
         return self._run_lock.locked()

tests/p2p/test_service.py

@@ -27,8 +27,10 @@ async def test_daemon_exit_causes_parent_cancellation():
     service = ParentService()
     asyncio.ensure_future(service.run())
     await asyncio.sleep(0.01)
+    assert service.daemon.is_operational
     assert service.daemon.is_running
     await service.daemon.cancel()
     await asyncio.sleep(0.01)
+    assert not service.is_operational
     assert not service.is_running
     await service.events.cleaned_up.wait()

tests/trinity/core/integration_test_helpers.py

@@ -16,7 +16,7 @@
 
 async def connect_to_peers_loop(peer_pool, nodes):
     """Loop forever trying to connect to one of the given nodes if the pool is not yet full."""
-    while not peer_pool.cancel_token.triggered:
+    while peer_pool.is_operational:
         try:
             if not peer_pool.is_full:
                 await peer_pool.connect_to_nodes(nodes)

trinity/plugins/builtin/tx_pool/pool.py

@@ -73,7 +73,7 @@ async def _run(self) -> None:
         self.logger.info("Running Tx Pool")
 
         with self.subscribe(self._peer_pool):
-            while not self.cancel_token.triggered:
+            while self.is_operational:
                 peer, cmd, msg = await self.wait(
                     self.msg_queue.get(), token=self.cancel_token)
                 peer = cast(ETHPeer, peer)

trinity/protocol/common/exchanges.py

@@ -60,7 +60,7 @@ async def get_result(
         - the manager service is running
         - the payload validator is primed with the request payload
         """
-        if not self._manager.is_running:
+        if not self._manager.is_operational:
             await self._manager.launch_service()
 
         # bind the outbound request payload to the payload validator

trinity/protocol/common/managers.py

@@ -118,7 +118,7 @@ async def _run(self) -> None:
         self.logger.debug("Launching %s for peer %s", self.__class__.__name__, self._peer)
 
         with self.subscribe_peer(self._peer):
-            while not self.cancel_token.triggered:
+            while self.is_operational:
                 peer, cmd, msg = await self.wait(self.msg_queue.get())
                 if peer != self._peer:
                     self.logger.error("Unexpected peer: %s  expected: %s", peer, self._peer)
@@ -202,8 +202,8 @@ async def launch_service(self) -> None:
         await self._response_stream.events.started.wait()
 
     @property
-    def is_running(self) -> bool:
-        return self.service is not None and self.service.is_running
+    def is_operational(self) -> bool:
+        return self.service is not None and self.service.is_operational
 
     async def get_result(
             self,
@@ -213,7 +213,7 @@ async def get_result(
             payload_validator: Callable[[TResponsePayload], None],
             timeout: int = None) -> TResult:
 
-        if not self.is_running:
+        if not self.is_operational:
             raise ValidationError("You must call `launch_service` before initiating a peer request")
 
         stream = self._response_stream