Ensure shared process plugins shutdown cleanly

Fixes #1284

Author: cburgdorf

setup.py

@@ -36,6 +36,7 @@
         "upnpclient>=0.0.8,<1",
     ],
     'trinity': [
+        "async-generator==1.10",
         "bloom-filter==1.3",
         "cachetools>=2.1.0,<3.0.0",
         "coincurve>=8.0.0,<9.0.0",

trinity/extensibility/__init__.py

@@ -2,15 +2,17 @@
     BaseEvent
 )
 from trinity.extensibility.plugin import (  # noqa: F401
-    BasePlugin,
+    BaseAsyncStopPlugin,
+    BaseMainProcessPlugin,
     BaseIsolatedPlugin,
+    BasePlugin,
+    BaseSyncStopPlugin,
     DebugPlugin,
     PluginContext,
-    PluginProcessScope,
 )
 from trinity.extensibility.plugin_manager import (  # noqa: F401
+    BaseManagerProcessScope,
     MainAndIsolatedProcessScope,
     PluginManager,
-    ManagerProcessScope,
     SharedProcessScope,
 )

trinity/extensibility/exceptions.py

@@ -0,0 +1,12 @@
+from trinity.exceptions import (
+    BaseTrinityError,
+)
+
+
+class UnsuitableShutdownError(BaseTrinityError):
+    """
+    Raised when `shutdown` was called on a ``PluginManager`` instance that operates
+    in the ``MainAndIsolatedProcessScope`` or when ``shutdown_nowait`` was called on a
+    ``PluginManager`` instance that operates in the ``SharedProcessScope``.
+    """
+    pass

trinity/extensibility/plugin.py

@@ -7,10 +7,7 @@
     Namespace,
     _SubParsersAction,
 )
-from enum import (
-    auto,
-    Enum,
-)
+import asyncio
 import logging
 from multiprocessing import (
     Process
@@ -48,20 +45,6 @@
 )
 
 
-class PluginProcessScope(Enum):
-    """
-    Define the process model in which a plugin operates:
-
-      - ISOLATED: The plugin runs in its own separate process
-      - MAIN: The plugin takes over the Trinity main process (e.g. attach)
-      - SHARED: The plugin runs in a process that is shared with other plugins
-    """
-
-    ISOLATED = auto()
-    MAIN = auto()
-    SHARED = auto()
-
-
 class PluginContext:
     """
     The ``PluginContext`` holds valuable contextual information such as the parsed
@@ -98,14 +81,6 @@ def name(self) -> str:
             "Must be implemented by subclasses"
         )
 
-    @property
-    def process_scope(self) -> PluginProcessScope:
-        """
-        Return the :class:`~trinity.extensibility.plugin.PluginProcessScope` that the plugin uses
-        to operate. The default scope is ``PluginProcessScope.SHARED``.
-        """
-        return PluginProcessScope.SHARED
-
     @property
     def logger(self) -> logging.Logger:
         return logging.getLogger('trinity.extensibility.plugin.BasePlugin#{0}'.format(self.name))
@@ -147,21 +122,42 @@ def start(self) -> None:
         """
         pass
 
+
+class BaseSyncStopPlugin(BasePlugin):
+    """
+    A ``BaseSyncStopPlugin`` unwinds synchronoulsy, hence blocks until shut down is done.
+    """
     def stop(self) -> None:
-        """
-        Called when the plugin gets stopped. Should be overwritten to perform cleanup
-        work in case the plugin set up external resources.
-        """
         pass
 
 
-class BaseIsolatedPlugin(BasePlugin):
+class BaseAsyncStopPlugin(BasePlugin):
+    """
+    A ``BaseAsyncStopPlugin`` unwinds asynchronoulsy, hence needs to be awaited.
+    """
 
-    _process: Process = None
+    async def stop(self) -> None:
+        pass
 
-    @property
-    def process_scope(self) -> PluginProcessScope:
-        return PluginProcessScope.ISOLATED
+
+class BaseMainProcessPlugin(BasePlugin):
+    """
+    A ``BaseMainProcessPlugin`` overtakes the whole main process before most of the Trinity boot
+    process had a chance to start. In that sense it redefines the whole meaning of the ``trinity``
+    process.
+    """
+    pass
+
+
+class BaseIsolatedPlugin(BaseSyncStopPlugin):
+    """
+    A ``BaseIsolatedPlugin`` runs in an isolated process and doesn't dictate whether its
+    implementation is based on non-blocking asyncio or synchronous calls. When an isolated
+    plugin is stopped it will first receive a SIGINT followed by a SIGTERM soon after.
+    It is up to the plugin to handle these signals accordingly.
+    """
+
+    _process: Process = None
 
     def _start(self) -> None:
         self._process = ctx.Process(
@@ -182,7 +178,7 @@ def stop(self) -> None:
         kill_process_gracefully(self._process, self.logger)
 
 
-class DebugPlugin(BasePlugin):
+class DebugPlugin(BaseAsyncStopPlugin):
     """
     This is a dummy plugin useful for demonstration and debugging purposes
     """
@@ -195,11 +191,22 @@ def configure_parser(self, arg_parser: ArgumentParser, subparser: _SubParsersAct
         arg_parser.add_argument("--debug-plugin", type=bool, required=False)
 
     def handle_event(self, activation_event: BaseEvent) -> None:
-        self.logger.info("Debug plugin: handle_event called: ", activation_event)
+        self.logger.info("Debug plugin: handle_event called: %s", activation_event)
 
     def should_start(self) -> bool:
         self.logger.info("Debug plugin: should_start called")
         return True
 
     def start(self) -> None:
         self.logger.info("Debug plugin: start called")
+        asyncio.ensure_future(self.count_forever())
+
+    async def count_forever(self) -> None:
+        i = 0
+        while True:
+            self.logger.info(i)
+            i += 1
+            await asyncio.sleep(1)
+
+    async def stop(self) -> None:
+        self.logger.info("Debug plugin: stop called")