Make the TxPool run as a plugin

Author: cburgdorf

tests/p2p/test_tx_pool.py

@@ -13,10 +13,10 @@
     force_bytes_to_address
 )
 
-from trinity.tx_pool.pool import (
+from trinity.plugins.builtin.tx_pool.pool import (
     TxPool,
 )
-from trinity.tx_pool.validators import (
+from trinity.plugins.builtin.tx_pool.validators import (
     DefaultTransactionValidator
 )
 

trinity/main.py

@@ -1,3 +1,4 @@
+from argparse import Namespace
 import asyncio
 import logging
 import signal
@@ -33,6 +34,15 @@
 from trinity.config import (
     ChainConfig,
 )
+from trinity.extensibility import (
+    PluginManager,
+)
+from trinity.extensibility.events import (
+    TrinityStartupEvent
+)
+from trinity.plugins.registry import (
+    ENABLED_PLUGINS
+)
 from trinity.utils.ipc import (
     wait_for_ipc,
     kill_process_gracefully,
@@ -81,6 +91,8 @@
 
 
 def main() -> None:
+    plugin_manager = setup_plugins()
+    plugin_manager.amend_argparser_config(parser)
     args = parser.parse_args()
 
     log_level = getattr(logging, args.log_level.upper())
@@ -154,7 +166,7 @@ def main() -> None:
 
     networking_process = ctx.Process(
         target=launch_node,
-        args=(chain_config, ),
+        args=(args, chain_config, ),
         kwargs=extra_kwargs,
     )
 
@@ -230,9 +242,21 @@ async def exit_on_signal(service_to_exit: BaseService) -> None:
 
 @setup_cprofiler('launch_node')
 @with_queued_logging
-def launch_node(chain_config: ChainConfig) -> None:
+def launch_node(args: Namespace, chain_config: ChainConfig) -> None:
     NodeClass = chain_config.node_class
-    node = NodeClass(chain_config)
+    # 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.
+    plugin_manager = setup_plugins()
+    plugin_manager.broadcast(TrinityStartupEvent(
+        args,
+        chain_config
+    ))
+
+    node = NodeClass(plugin_manager, chain_config)
 
     run_service_until_quit(node)
 
@@ -250,3 +274,11 @@ def run_service_until_quit(service: BaseService) -> None:
     asyncio.ensure_future(service.run())
     loop.run_forever()
     loop.close()
+
+
+def setup_plugins() -> PluginManager:
+    plugin_manager = PluginManager()
+    # TODO: Implement auto-discovery of plugins based on some convention/configuration scheme
+    plugin_manager.register(ENABLED_PLUGINS)
+
+    return plugin_manager

trinity/nodes/base.py

@@ -41,11 +41,11 @@
 from trinity.config import (
     ChainConfig,
 )
-from trinity.tx_pool.pool import (
-    TxPool
+from trinity.extensibility import (
+    PluginManager,
 )
-from trinity.tx_pool.validators import (
-    DefaultTransactionValidator
+from trinity.extensibility.events import (
+    ResourceAvailableEvent
 )
 
 
@@ -55,11 +55,10 @@ class Node(BaseService):
     unset attributes.
     """
     chain_class: Type[BaseChain] = None
-    initial_tx_validation_block_number: int = None
 
-    def __init__(self, chain_config: ChainConfig) -> None:
+    def __init__(self, plugin_manager: PluginManager, chain_config: ChainConfig) -> None:
         super().__init__()
-
+        self._plugin_manager = plugin_manager
         self._db_manager = create_db_manager(chain_config.database_ipc_path)
         self._db_manager.connect()  # type: ignore
         self._headerdb = self._db_manager.get_headerdb()  # type: ignore
@@ -100,13 +99,22 @@ def add_service(self, service: BaseService) -> None:
         else:
             self._auxiliary_services.append(service)
 
-    def create_and_add_tx_pool(self) -> None:
-        self.tx_pool = TxPool(
-            self.get_peer_pool(),
-            DefaultTransactionValidator(self.get_chain(), self.initial_tx_validation_block_number),
-            self.cancel_token
-        )
-        self.add_service(self.tx_pool)
+    def notify_resource_available(self) -> None:
+
+        # We currently need this to give plugins the chance to start as soon
+        # as the `PeerPool` is available. In the long term, the peer pool may become
+        # a plugin itself and we can get rid of this.
+        self._plugin_manager.broadcast(ResourceAvailableEvent(
+            resource=(self.get_peer_pool(), self.cancel_token),
+            resource_type=PeerPool
+        ))
+
+        # This broadcasts the *local* chain, which is suited for tasks that aren't blocking
+        # for too long. There may be value in also broadcasting the proxied chain.
+        self._plugin_manager.broadcast(ResourceAvailableEvent(
+            resource=self.get_chain(),
+            resource_type=BaseChain
+        ))
 
     def make_ipc_server(self) -> Union[IPCServer, EmptyService]:
         if self._jsonrpc_ipc_path:

trinity/nodes/full.py

@@ -13,21 +13,24 @@
 from trinity.config import (
     ChainConfig,
 )
+from trinity.extensibility import (
+    PluginManager
+)
 
 
 class FullNode(Node):
     _chain: BaseChain = None
     _p2p_server: Server = None
 
-    def __init__(self, chain_config: ChainConfig) -> None:
-        super().__init__(chain_config)
+    def __init__(self, plugin_manager: PluginManager, chain_config: ChainConfig) -> None:
+        super().__init__(plugin_manager, chain_config)
         self._bootstrap_nodes = chain_config.bootstrap_nodes
         self._preferred_nodes = chain_config.preferred_nodes
         self._network_id = chain_config.network_id
         self._node_key = chain_config.nodekey
         self._node_port = chain_config.port
         self._max_peers = chain_config.max_peers
-        self.create_and_add_tx_pool()
+        self.notify_resource_available()
 
     def get_chain(self) -> BaseChain:
         if self._chain is None:

trinity/nodes/light.py

@@ -15,6 +15,9 @@
 from trinity.chains.light import (
     LightDispatchChain,
 )
+from trinity.extensibility import (
+    PluginManager
+)
 from trinity.nodes.base import Node
 from trinity.config import (
     ChainConfig,
@@ -31,8 +34,8 @@ class LightNode(Node):
     network_id: int = None
     nodekey: PrivateKey = None
 
-    def __init__(self, chain_config: ChainConfig) -> None:
-        super().__init__(chain_config)
+    def __init__(self, plugin_manager: PluginManager, chain_config: ChainConfig) -> None:
+        super().__init__(plugin_manager, chain_config)
 
         self.network_id = chain_config.network_id
         self.nodekey = chain_config.nodekey
@@ -51,7 +54,7 @@ def __init__(self, chain_config: ChainConfig) -> None:
         self.add_service(self._discovery)
         self.add_service(self._peer_pool)
         self.add_service(self._peer_chain)
-        self.create_and_add_tx_pool()
+        self.notify_resource_available()
 
     async def _run(self) -> None:
         # TODO add a datagram endpoint service that can be added with self.add_service

trinity/nodes/mainnet.py

@@ -1,6 +1,5 @@
 from evm.chains.mainnet import (
     MainnetChain,
-    BYZANTIUM_MAINNET_BLOCK
 )
 
 from trinity.chains.mainnet import (
@@ -12,9 +11,7 @@
 
 class MainnetFullNode(FullNode):
     chain_class = MainnetChain
-    initial_tx_validation_block_number = BYZANTIUM_MAINNET_BLOCK
 
 
 class MainnetLightNode(LightNode):
     chain_class = MainnetLightDispatchChain
-    initial_tx_validation_block_number = BYZANTIUM_MAINNET_BLOCK

trinity/nodes/ropsten.py

@@ -1,6 +1,5 @@
 from evm.chains.ropsten import (
     RopstenChain,
-    BYZANTIUM_ROPSTEN_BLOCK
 )
 
 from trinity.chains.ropsten import (
@@ -12,9 +11,7 @@
 
 class RopstenFullNode(FullNode):
     chain_class = RopstenChain
-    initial_tx_validation_block_number = BYZANTIUM_ROPSTEN_BLOCK
 
 
 class RopstenLightNode(LightNode):
     chain_class = RopstenLightDispatchChain
-    initial_tx_validation_block_number = BYZANTIUM_ROPSTEN_BLOCK