Merge pull request #1245 from carver/service-exceptions

Clearer service exceptions

Author: carver

Makefile

@@ -23,7 +23,7 @@ clean-pyc:
 	find . -name '*~' -exec rm -f {} +
 
 lint:
-	tox -elint-py3{6,5}
+	tox -epy3{6,5}-lint
 
 test:
 	py.test --tb native tests

p2p/peer.py

@@ -732,7 +732,7 @@ def unsubscribe(self, subscriber: PeerSubscriber) -> None:
             peer.remove_subscriber(subscriber)
 
     async def start_peer(self, peer: BasePeer) -> None:
-        self.run_task(peer.run())
+        self.run_child_service(peer)
         await self.wait(peer.events.started.wait(), timeout=1)
         try:
             # Although connect() may seem like a more appropriate place to perform the DAO fork

p2p/service.py

@@ -8,14 +8,16 @@
     Callable,
     List,
     Optional,
-    Set,
     cast,
 )
 from weakref import WeakSet
 
-from eth.tools.logging import TraceLogger
-
 from cancel_token import CancelToken, OperationCancelled
+from eth_utils import (
+    ValidationError,
+)
+
+from eth.tools.logging import TraceLogger
 
 from p2p.cancellable import CancellableMixin
 from p2p.utils import get_asyncio_executor
@@ -32,8 +34,8 @@ def __init__(self) -> None:
 
 class BaseService(ABC, CancellableMixin):
     logger: TraceLogger = None
-    _child_services: Set['BaseService']
     # Use a WeakSet so that we don't have to bother updating it when tasks finish.
+    _child_services: 'WeakSet[BaseService]'
     _tasks: 'WeakSet[asyncio.Future[Any]]'
     _finished_callbacks: List[Callable[['BaseService'], None]]
     # Number of seconds cancel() will wait for run() to finish.
@@ -49,7 +51,7 @@ def __init__(self,
                  loop: asyncio.AbstractEventLoop = None) -> None:
         self.events = ServiceEvents()
         self._run_lock = asyncio.Lock()
-        self._child_services = set()
+        self._child_services = WeakSet()
         self._tasks = WeakSet()
         self._finished_callbacks = []
 
@@ -88,9 +90,9 @@ async def run(
         finished_callback (if one was passed).
         """
         if self.is_running:
-            raise RuntimeError("Cannot start the service while it's already running")
+            raise ValidationError("Cannot start the service while it's already running")
         elif self.is_cancelled:
-            raise RuntimeError("Cannot restart a service that has already been cancelled")
+            raise ValidationError("Cannot restart a service that has already been cancelled")
 
         if finished_callback:
             self._finished_callbacks.append(finished_callback)
@@ -144,6 +146,15 @@ def run_child_service(self, child_service: 'BaseService') -> None:
         """
         Run a child service and keep a reference to it to be considered during the cleanup.
         """
+        if child_service.is_running:
+            raise ValidationError(
+                f"Can't start service {child_service!r}, child of {self!r}: it's already running"
+            )
+        elif child_service.is_cancelled:
+            raise ValidationError(
+                f"Can't restart {child_service!r}, child of {self!r}: it's already completed"
+            )
+
         self._child_services.add(child_service)
         self.run_task(child_service.run())
 
@@ -153,6 +164,15 @@ def run_daemon(self, service: 'BaseService') -> None:
 
         If the service finishes while we're still running, we'll terminate as well.
         """
+        if service.is_running:
+            raise ValidationError(
+                f"Can't start daemon {service!r}, child of {self!r}: it's already running"
+            )
+        elif service.is_cancelled:
+            raise ValidationError(
+                f"Can't restart daemon {service!r}, child of {self!r}: it's already completed"
+            )
+
         self._child_services.add(service)
 
         async def _run_daemon_wrapper() -> None:
@@ -193,7 +213,7 @@ async def cancel(self) -> None:
             self.logger.warning("Tried to cancel %s, but it was already cancelled", self)
             return
         elif not self.is_running:
-            raise RuntimeError("Cannot cancel a service that has not been started")
+            raise ValidationError("Cannot cancel a service that has not been started")
 
         self.logger.debug("Cancelling %s", self)
         self.events.cancelled.set()

tests/p2p/test_service.py

@@ -39,3 +39,60 @@ async def test_daemon_exit_causes_parent_cancellation():
     assert not service.is_running
 
     await asyncio.wait_for(service.events.cleaned_up.wait(), timeout=1)
+
+
+@pytest.mark.asyncio
+async def test_service_tasks_do_not_leak_memory():
+    service = WaitService()
+    asyncio.ensure_future(service.run())
+
+    end = asyncio.Event()
+
+    async def run_until_end():
+        await end.wait()
+
+    service.run_task(run_until_end())
+
+    # inspect internals to determine if memory is leaking
+
+    # confirm that task is tracked:
+    assert len(service._tasks) == 1
+
+    end.set()
+    # allow the coro to exit
+    await asyncio.sleep(0)
+
+    # confirm that task is no longer tracked:
+    assert len(service._tasks) == 0
+
+    # test cleanup
+    await service.cancel()
+
+
+@pytest.mark.asyncio
+async def test_service_children_do_not_leak_memory():
+    parent = WaitService()
+    child = WaitService()
+    asyncio.ensure_future(parent.run())
+
+    parent.run_child_service(child)
+
+    # inspect internals to determine if memory is leaking
+
+    # confirm that child service is tracked:
+    assert len(parent._child_services) == 1
+
+    # give child a chance to start
+    await asyncio.sleep(0)
+
+    # ... and then end it
+    await child.cancel()
+
+    # remove the final reference to the child service
+    del child
+
+    # confirm that child service is no longer tracked:
+    assert len(parent._child_services) == 0
+
+    # test cleanup
+    await parent.cancel()

tox.ini

@@ -6,7 +6,7 @@ envlist=
     py{36}-rpc-state-{frontier,homestead,eip150,eip158,byzantium,quadratic}
     py{35,36}-native-state-{frontier,homestead,eip150,eip158,byzantium,constantinople,metropolis}
     py37-{core,trinity,trinity-integration}
-    lint-py{35,36}
+    py{35,36}-lint
 
 [flake8]
 max-line-length= 100