Ensure isolated plugins are stopped concurrently

So, far isolated plugins were taken down sequentially.
The more trinity moves towards a multi process architecture,
the more this is likely to become a problem.

Taking down these processes concurrently is a bit of
a challenge because the code runs under the asumption
that the event loop may or may not be closed already.

Hence, this code tries to avoid introducing async
for the sake of isolated process take down.

Author: cburgdorf

docs/api/trinity/extensibility/api.extensibility.plugin.rst

@@ -14,12 +14,6 @@ BasePlugin
 .. autoclass:: trinity.extensibility.plugin.BasePlugin
   :members:
 
-BaseSyncStopPlugin
-------------------
-
-.. autoclass:: trinity.extensibility.plugin.BaseSyncStopPlugin
-  :members:
-
 BaseAsyncStopPlugin
 -------------------
 

trinity/extensibility/__init__.py

@@ -6,7 +6,6 @@
     BaseMainProcessPlugin,
     BaseIsolatedPlugin,
     BasePlugin,
-    BaseSyncStopPlugin,
     DebugPlugin,
     PluginContext,
     TrinityBootInfo,

trinity/extensibility/plugin.py

@@ -44,9 +44,6 @@
     EventBusNotReady,
     InvalidPluginStatus,
 )
-from trinity.utils.ipc import (
-    kill_process_gracefully
-)
 from trinity.utils.mp import (
     ctx,
 )
@@ -229,26 +226,6 @@ def do_start(self) -> None:
         pass
 
 
-class BaseSyncStopPlugin(BasePlugin):
-    """
-    A :class:`~trinity.extensibility.plugin.BaseSyncStopPlugin` unwinds synchronoulsy, hence blocks
-    until the shutdown is done.
-    """
-    def do_stop(self) -> None:
-        """
-        Stop the plugin. Should be overwritten by subclasses.
-        """
-        pass
-
-    def stop(self) -> None:
-        """
-        Delegate to :meth:`~trinity.extensibility.plugin.BaseSyncStopPlugin.do_stop` causing the
-        plugin to stop and setting ``running`` to ``False``.
-        """
-        self.do_stop()
-        self._status = PluginStatus.STOPPED
-
-
 class BaseAsyncStopPlugin(BasePlugin):
     """
     A :class:`~trinity.extensibility.plugin.BaseAsyncStopPlugin` unwinds asynchronoulsy, hence
@@ -279,7 +256,7 @@ class BaseMainProcessPlugin(BasePlugin):
     pass
 
 
-class BaseIsolatedPlugin(BaseSyncStopPlugin):
+class BaseIsolatedPlugin(BasePlugin):
     """
     A :class:`~trinity.extensibility.plugin.BaseIsolatedPlugin` runs in an isolated process and
     hence provides security and flexibility by not making assumptions about its internal
@@ -292,6 +269,13 @@ class BaseIsolatedPlugin(BaseSyncStopPlugin):
 
     _process: Process = None
 
+    @property
+    def process(self) -> Process:
+        """
+        Return the ``Process`` created by the isolated plugin.
+        """
+        return self._process
+
     def start(self) -> None:
         """
         Prepare the plugin to get started and eventually call ``do_start`` in a separate process.
@@ -314,8 +298,14 @@ def _prepare_start(self) -> None:
         )
         self.do_start()
 
-    def do_stop(self) -> None:
-        kill_process_gracefully(self._process, self.logger)
+    def stop(self) -> None:
+        """
+        Set the ``status`` to `STOPPED`` but rely on the
+        :class:`~trinity.extensibility.plugin_manager.PluginManager` to tear down the process. This
+        allows isolated plugins to be taken down concurrently without depending on a running
+        event loop.
+        """
+        self._status = PluginStatus.STOPPED
 
 
 class DebugPlugin(BaseAsyncStopPlugin):

trinity/extensibility/plugin_manager.py

@@ -35,10 +35,12 @@
     BaseIsolatedPlugin,
     BaseMainProcessPlugin,
     BasePlugin,
-    BaseSyncStopPlugin,
     PluginContext,
     TrinityBootInfo,
 )
+from trinity.utils.ipc import (
+    kill_processes_gracefully,
+)
 
 
 class BaseManagerProcessScope(ABC):
@@ -215,17 +217,19 @@ def shutdown_blocking(self) -> None:
 
         self._logger.info("Shutting down PluginManager with scope %s", type(self._scope))
 
-        for plugin in self._plugin_store:
+        plugins = [
+            plugin for plugin in self._plugin_store
+            if isinstance(plugin, BaseIsolatedPlugin) and plugin.running
+        ]
+        processes = [plugin.process for plugin in plugins]
 
-            if not isinstance(plugin, BaseSyncStopPlugin) or not plugin.running:
-                continue
+        for plugin in plugins:
+            self._logger.info("Stopping plugin: %s", plugin.name)
 
-            try:
-                self._logger.info("Stopping plugin: %s", plugin.name)
-                plugin.stop()
-                self._logger.info("Successfully stopped plugin: %s", plugin.name)
-            except Exception:
-                self._logger.exception("Exception thrown while stopping plugin %s", plugin.name)
+        kill_processes_gracefully(processes, self._logger)
+
+        for plugin in plugins:
+            self._logger.info("Successfully stopped plugin: %s", plugin.name)
 
     async def shutdown(self) -> None:
         """

trinity/utils/ipc.py

@@ -5,7 +5,7 @@
 import signal
 import subprocess
 import time
-from typing import Callable
+from typing import Callable, Iterable
 
 
 def wait_for_ipc(ipc_path: pathlib.Path, timeout: int=10) -> None:
@@ -35,6 +35,42 @@ def kill_process_gracefully(
     kill_process_id_gracefully(process.pid, process.join, logger, SIGINT_timeout, SIGTERM_timeout)
 
 
+def kill_processes_gracefully(
+        processes: Iterable[Process],
+        logger: Logger,
+        SIGINT_timeout: int=DEFAULT_SIGINT_TIMEOUT,
+        SIGTERM_timeout: int=DEFAULT_SIGTERM_TIMEOUT) -> None:
+
+    # Send SIGINT to each process without blocking
+    for process in processes:
+        sigint_process_id(process.pid, lambda _: None, logger, SIGINT_timeout)
+
+    # Now block on each process as long as we have time left in the budget
+    sigint_at = time.time()
+    for process in processes:
+        waited_sec = time.time() - sigint_at
+        if waited_sec >= SIGINT_timeout:
+            logger.debug("Waited %d on SIGINT, moving on", waited_sec)
+            break
+        process.join(SIGINT_timeout)
+
+    # Send SIGTERM to each process without blocking
+    for process in processes:
+        sigterm_process_id(process.pid, lambda _: None, logger, SIGTERM_timeout)
+
+    # Now block on each process as long as we have time left in the budget
+    sigterm_at = time.time()
+    for process in processes:
+        waited_sec = time.time() - sigterm_at
+        if waited_sec >= SIGTERM_timeout:
+            logger.debug("Waited %d on SIGINT, moving on", waited_sec)
+            break
+        process.join(SIGTERM_timeout)
+
+    for process in processes:
+        sigkill_process_id(process.pid, logger)
+
+
 def kill_popen_gracefully(
         popen: subprocess.Popen,
         logger: Logger,
@@ -56,6 +92,18 @@ def kill_process_id_gracefully(
         logger: Logger,
         SIGINT_timeout: int=DEFAULT_SIGINT_TIMEOUT,
         SIGTERM_timeout: int=DEFAULT_SIGTERM_TIMEOUT) -> None:
+
+    sigint_process_id(process_id, wait_for_completion, logger, SIGINT_timeout)
+    sigterm_process_id(process_id, wait_for_completion, logger, SIGTERM_timeout)
+    sigkill_process_id(process_id, logger)
+
+
+def sigint_process_id(
+        process_id: int,
+        wait_for_completion: Callable[[int], None],
+        logger: Logger,
+        SIGINT_timeout: int=DEFAULT_SIGINT_TIMEOUT) -> None:
+
     try:
         try:
             os.kill(process_id, signal.SIGINT)
@@ -72,6 +120,13 @@ def kill_process_id_gracefully(
             "with CTRL+C two more times."
         )
 
+
+def sigterm_process_id(
+        process_id: int,
+        wait_for_completion: Callable[[int], None],
+        logger: Logger,
+        SIGTERM_timeout: int=DEFAULT_SIGTERM_TIMEOUT) -> None:
+
     try:
         try:
             os.kill(process_id, signal.SIGTERM)
@@ -88,6 +143,11 @@ def kill_process_id_gracefully(
             "with CTRL+C one more time."
         )
 
+
+def sigkill_process_id(
+        process_id: int,
+        logger: Logger) -> None:
+
     try:
         os.kill(process_id, signal.SIGKILL)
     except ProcessLookupError: