Refactor plugin architecture

Author: cburgdorf

setup.py

@@ -43,7 +43,7 @@
         "ipython>=6.2.1,<7.0.0",
         "plyvel==1.0.5",
         "web3==4.4.1",
-        "lahja==0.8.0",
+        "lahja==0.9.0",
         "uvloop==0.11.2;platform_system=='Linux' or platform_system=='Darwin'",
         "websockets==5.0.1",
     ],

tests/trinity/conftest.py

@@ -65,11 +65,11 @@ def event_loop():
 
 
 @pytest.fixture(scope='module')
-def event_bus(event_loop):
+async def event_bus(event_loop):
     bus = EventBus()
     endpoint = bus.create_endpoint('test')
     bus.start(event_loop)
-    endpoint.connect(event_loop)
+    await endpoint.connect(event_loop)
     try:
         yield endpoint
     finally:

trinity/extensibility/events.py

@@ -3,12 +3,9 @@
     Type,
     TYPE_CHECKING,
 )
-from argparse import (
-    Namespace,
-)
 
-from trinity.config import (
-    TrinityConfig,
+from lahja import (
+    BaseEvent,
 )
 
 
@@ -18,23 +15,12 @@
     )
 
 
-class BaseEvent:
-    """
-    The base class for all plugin events. Plugin events can be broadcasted for all different
-    kind of reasons. Plugins can act based on these events and consume the events even before
-    the plugin is started, giving plugins the chance to start based on an event or a series of
-    events. The startup of Trinity itself can be an event as well as the start of a plugin itself
-    which, for instance, gives other plugins the chance to start based on these previous events.
-    """
-    pass
-
-
 class PluginStartedEvent(BaseEvent):
     """
     Broadcasted when a plugin was started
     """
-    def __init__(self, plugin: 'BasePlugin') -> None:
-        self.plugin = plugin
+    def __init__(self, plugin_type: Type['BasePlugin']) -> None:
+        self.plugin_type = plugin_type
 
 
 class ResourceAvailableEvent(BaseEvent):

trinity/extensibility/exceptions.py

@@ -10,3 +10,11 @@ class UnsuitableShutdownError(BaseTrinityError):
     ``PluginManager`` instance that operates in the ``SharedProcessScope``.
     """
     pass
+
+
+class EventBusNotReady(BaseTrinityError):
+    """
+    Raised when a plugin tried to access an EventBus before the plugin
+    had received its ``ready`` call.
+    """
+    pass

trinity/extensibility/plugin.py

@@ -29,10 +29,14 @@
     MAIN_EVENTBUS_ENDPOINT
 )
 from trinity.events import (
-    ShutdownRequest
+    ShutdownRequest,
 )
 from trinity.extensibility.events import (
-    BaseEvent
+    BaseEvent,
+    PluginStartedEvent,
+)
+from trinity.extensibility.exceptions import (
+    EventBusNotReady,
 )
 from trinity.utils.ipc import (
     kill_process_gracefully
@@ -70,6 +74,7 @@ def shutdown_host(self) -> None:
 class BasePlugin(ABC):
 
     context: PluginContext = None
+    running: bool = False
 
     @property
     @abstractmethod
@@ -85,35 +90,41 @@ def name(self) -> str:
     def logger(self) -> logging.Logger:
         return logging.getLogger('trinity.extensibility.plugin.BasePlugin#{0}'.format(self.name))
 
+    @property
+    def event_bus(self) -> Endpoint:
+        if self.context is None:
+            raise EventBusNotReady("Tried accessing ``event_bus`` before ``ready`` was called")
+
+        return self.context.event_bus
+
     def set_context(self, context: PluginContext) -> None:
         """
         Set the :class:`~trinity.extensibility.plugin.PluginContext` for this plugin.
         """
         self.context = context
 
-    def configure_parser(self, arg_parser: ArgumentParser, subparser: _SubParsersAction) -> None:
+    def ready(self) -> None:
         """
-        Called at startup, giving the plugin a chance to amend the Trinity CLI argument parser
+        Called after the plugin received its context and is ready to bootstrap itself.
         """
         pass
 
-    def handle_event(self, activation_event: BaseEvent) -> None:
+    def configure_parser(self, arg_parser: ArgumentParser, subparser: _SubParsersAction) -> None:
         """
-        Notify the plugin about an event, giving it the chance to do internal accounting right
-        before :meth:`~trinity.extensibility.plugin.BasePlugin.should_start` is called
+        Called at startup, giving the plugin a chance to amend the Trinity CLI argument parser
         """
-
         pass
 
-    def should_start(self) -> bool:
+    def boot(self) -> None:
         """
-        Return ``True`` if the plugin should start, otherwise return ``False``
+        Prepare the plugin to get started and eventually cause ``start`` to get called.
         """
-
-        return False
-
-    def _start(self) -> None:
+        self.running = True
         self.start()
+        self.event_bus.broadcast(
+            PluginStartedEvent(type(self))
+        )
+        self.logger.info("Plugin started: %s", self.name)
 
     def start(self) -> None:
         """
@@ -159,18 +170,26 @@ class BaseIsolatedPlugin(BaseSyncStopPlugin):
 
     _process: Process = None
 
-    def _start(self) -> None:
+    def boot(self) -> None:
+        """
+        Prepare the plugin to get started and eventually cause ``start`` to get called.
+        """
+        self.running = True
         self._process = ctx.Process(
             target=self._prepare_start,
         )
 
         self._process.start()
+        self.logger.info("Plugin started: %s", self.name)
 
     def _prepare_start(self) -> None:
         log_queue = self.context.boot_kwargs['log_queue']
         level = self.context.boot_kwargs.get('log_level', logging.INFO)
         setup_queue_logging(log_queue, level)
-
+        self.event_bus.connect_no_wait()
+        self.event_bus.broadcast(
+            PluginStartedEvent(type(self))
+        )
         self.start()
 
     def stop(self) -> None:
@@ -193,10 +212,6 @@ def configure_parser(self, arg_parser: ArgumentParser, subparser: _SubParsersAct
     def handle_event(self, activation_event: BaseEvent) -> None:
         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())

trinity/extensibility/plugin_manager.py

@@ -27,10 +27,6 @@
 from trinity.config import (
     TrinityConfig
 )
-from trinity.extensibility.events import (
-    BaseEvent,
-    PluginStartedEvent,
-)
 from trinity.extensibility.exceptions import (
     UnsuitableShutdownError,
 )
@@ -136,7 +132,6 @@ class PluginManager:
     def __init__(self, scope: BaseManagerProcessScope) -> None:
         self._scope = scope
         self._plugin_store: List[BasePlugin] = []
-        self._started_plugins: List[BasePlugin] = []
         self._logger = logging.getLogger("trinity.extensibility.plugin_manager.PluginManager")
 
     @property
@@ -165,34 +160,6 @@ def amend_argparser_config(self,
         for plugin in self._plugin_store:
             plugin.configure_parser(arg_parser, subparser)
 
-    def broadcast(self, event: BaseEvent, exclude: BasePlugin = None) -> None:
-        """
-        Notify every registered :class:`~trinity.extensibility.plugin.BasePlugin` about an
-        event and check whether the plugin wants to start based on that event.
-
-        If a plugin gets started it will cause a
-        :class:`~trinity.extensibility.events.PluginStartedEvent` to get
-        broadcasted to all other plugins, giving them the chance to start based on that.
-        """
-        for plugin in self._plugin_store:
-
-            if plugin is exclude or not self._scope.is_responsible_for_plugin(plugin):
-                self._logger.debug("Skipping plugin %s (not responsible)", plugin.name)
-                continue
-
-            plugin.handle_event(event)
-
-            if plugin in self._started_plugins:
-                continue
-
-            if not plugin.should_start():
-                continue
-
-            plugin._start()
-            self._started_plugins.append(plugin)
-            self._logger.info("Plugin started: %s", plugin.name)
-            self.broadcast(PluginStartedEvent(plugin), plugin)
-
     def prepare(self,
                 args: Namespace,
                 trinity_config: TrinityConfig,
@@ -208,6 +175,7 @@ def prepare(self,
 
             context = self._scope.create_plugin_context(plugin, args, trinity_config, boot_kwargs)
             plugin.set_context(context)
+            plugin.ready()
 
     def shutdown_blocking(self) -> None:
         """
@@ -219,14 +187,15 @@ def shutdown_blocking(self) -> None:
 
         self._logger.info("Shutting down PluginManager with scope %s", type(self._scope))
 
-        for plugin in self._started_plugins:
+        for plugin in self._plugin_store:
 
-            if not isinstance(plugin, BaseSyncStopPlugin):
+            if not isinstance(plugin, BaseSyncStopPlugin) or not plugin.running:
                 continue
 
             try:
                 self._logger.info("Stopping plugin: %s", plugin.name)
                 plugin.stop()
+                plugin.running = False
                 self._logger.info("Successfully stopped plugin: %s", plugin.name)
             except Exception:
                 self._logger.exception("Exception thrown while stopping plugin %s", plugin.name)
@@ -241,8 +210,8 @@ async def shutdown(self) -> None:
         self._logger.info("Shutting down PluginManager with scope %s", type(self._scope))
 
         async_plugins = [
-            plugin for plugin in self._started_plugins
-            if isinstance(plugin, BaseAsyncStopPlugin)
+            plugin for plugin in self._plugin_store
+            if isinstance(plugin, BaseAsyncStopPlugin) and plugin.running
         ]
 
         stop_results = await asyncio.gather(
@@ -255,6 +224,7 @@ async def shutdown(self) -> None:
                     'Exception thrown while stopping plugin %s: %s', plugin.name, result
                 )
             else:
+                plugin.running = False
                 self._logger.info("Successfully stopped plugin: %s", plugin.name)
 
     def _stop_plugins(self,

trinity/main.py

@@ -108,7 +108,7 @@
 def main() -> None:
     event_bus = EventBus(ctx)
     main_endpoint = event_bus.create_endpoint(MAIN_EVENTBUS_ENDPOINT)
-    main_endpoint.connect()
+    main_endpoint.connect_no_wait()
 
     plugin_manager = setup_plugins(
         MainAndIsolatedProcessScope(event_bus, main_endpoint)
@@ -343,21 +343,15 @@ def _sigint_handler(*args: Any) -> None:
 def launch_node(args: Namespace, trinity_config: TrinityConfig, endpoint: Endpoint) -> None:
     with trinity_config.process_id_file('networking'):
 
-        endpoint.connect()
-
         NodeClass = trinity_config.node_class
-        # Temporary hack: We setup a second instance of the PluginManager.
-        # The first instance was only to configure the ArgumentParser whereas
-        # for now, the second instance that lives inside the networking process
-        # performs the bulk of the work. In the future, the PluginManager
-        # should probably live in its own process and manage whether plugins
-        # run in the shared plugin process or spawn their own.
+        node = NodeClass(endpoint, trinity_config)
+        loop = node.get_event_loop()
 
+        endpoint.connect_no_wait(loop)
+        # This is a second PluginManager instance governing plugins in a shared process.
         plugin_manager = setup_plugins(SharedProcessScope(endpoint))
         plugin_manager.prepare(args, trinity_config)
 
-        node = NodeClass(plugin_manager, trinity_config)
-        loop = node.get_event_loop()
         asyncio.ensure_future(handle_networking_exit(node, plugin_manager, endpoint), loop=loop)
         asyncio.ensure_future(node.run(), loop=loop)
         loop.run_forever()