Merge pull request #1353 from carver/decouple-handlers-from-peer-watcher

Separate Peer Message Handling from Syncing

Author: carver

tests/trinity/core/p2p-proto/test_server.py

@@ -61,6 +61,9 @@ def _make_peer_pool(self):
     def _make_syncer(self):
         return
 
+    def _make_request_server(self):
+        return
+
 
 def get_server(privkey, address):
     base_db = AtomicDB()

tests/trinity/core/p2p-proto/test_sync.py

@@ -10,7 +10,9 @@
 from eth.vm.forks.frontier import FrontierVM
 
 
+from trinity.protocol.eth.servers import ETHRequestServer
 from trinity.protocol.les.peer import LESPeer
+from trinity.protocol.les.servers import LightRequestServer
 from trinity.sync.full.chain import FastChainSyncer, RegularChainSyncer
 from trinity.sync.full.state import StateDownloader
 from trinity.sync.light.chain import LightChainSyncer
@@ -45,11 +47,17 @@ async def test_fast_syncer(request, event_loop, chaindb_fresh, chaindb_20):
         bob_headerdb=FakeAsyncHeaderDB(chaindb_20.db))
     client_peer_pool = MockPeerPoolWithConnectedPeers([client_peer])
     client = FastChainSyncer(FrontierTestChain(chaindb_fresh.db), chaindb_fresh, client_peer_pool)
+    server_peer_pool = MockPeerPoolWithConnectedPeers([server_peer])
     server = RegularChainSyncer(
         FrontierTestChain(chaindb_20.db),
         chaindb_20,
-        MockPeerPoolWithConnectedPeers([server_peer]))
+        server_peer_pool,
+    )
     asyncio.ensure_future(server.run())
+    server_request_handler = ETHRequestServer(FakeAsyncChainDB(chaindb_20.db), server_peer_pool)
+    asyncio.ensure_future(server_request_handler.run())
+    server_peer.logger.info("%s is serving 20 blocks", server_peer)
+    client_peer.logger.info("%s is syncing up 20", client_peer)
 
     def finalizer():
         event_loop.run_until_complete(server.cancel())
@@ -81,11 +89,17 @@ async def test_regular_syncer(request, event_loop, chaindb_fresh, chaindb_20):
         FrontierTestChain(chaindb_fresh.db),
         chaindb_fresh,
         MockPeerPoolWithConnectedPeers([client_peer]))
+    server_peer_pool = MockPeerPoolWithConnectedPeers([server_peer])
     server = RegularChainSyncer(
         FrontierTestChain(chaindb_20.db),
         chaindb_20,
-        MockPeerPoolWithConnectedPeers([server_peer]))
+        server_peer_pool,
+    )
     asyncio.ensure_future(server.run())
+    server_request_handler = ETHRequestServer(FakeAsyncChainDB(chaindb_20.db), server_peer_pool)
+    asyncio.ensure_future(server_request_handler.run())
+    server_peer.logger.info("%s is serving 20 blocks", server_peer)
+    client_peer.logger.info("%s is syncing up 20", client_peer)
 
     def finalizer():
         event_loop.run_until_complete(asyncio.gather(
@@ -115,11 +129,17 @@ async def test_light_syncer(request, event_loop, chaindb_fresh, chaindb_20):
         FrontierTestChain(chaindb_fresh.db),
         chaindb_fresh,
         MockPeerPoolWithConnectedPeers([client_peer]))
+    server_peer_pool = MockPeerPoolWithConnectedPeers([server_peer])
     server = LightChainSyncer(
         FrontierTestChain(chaindb_20.db),
         chaindb_20,
-        MockPeerPoolWithConnectedPeers([server_peer]))
+        server_peer_pool,
+    )
     asyncio.ensure_future(server.run())
+    server_request_handler = LightRequestServer(FakeAsyncHeaderDB(chaindb_20.db), server_peer_pool)
+    asyncio.ensure_future(server_request_handler.run())
+    server_peer.logger.info("%s is serving 20 blocks", server_peer)
+    client_peer.logger.info("%s is syncing up 20", client_peer)
 
     def finalizer():
         event_loop.run_until_complete(asyncio.gather(

tests/trinity/integration/test_lightchain_integration.py

@@ -27,6 +27,7 @@
 from trinity.constants import ROPSTEN_NETWORK_ID
 from trinity.protocol.common.context import ChainContext
 from trinity.protocol.les.peer import LESPeerPool
+from trinity.protocol.les.servers import LightRequestServer
 from trinity.sync.light.chain import LightChainSyncer
 from trinity.sync.light.service import LightPeerChain
 from trinity.utils.ipc import (
@@ -176,17 +177,20 @@ async def test_lightchain_integration(
     syncer = LightChainSyncer(chain, chaindb, peer_pool)
     syncer.min_peers_to_sync = 1
     peer_chain = LightPeerChain(headerdb, peer_pool)
+    server_request_handler = LightRequestServer(headerdb, peer_pool)
 
     asyncio.ensure_future(peer_pool.run())
     asyncio.ensure_future(connect_to_peers_loop(peer_pool, tuple([remote])))
     asyncio.ensure_future(peer_chain.run())
+    asyncio.ensure_future(server_request_handler.run())
     asyncio.ensure_future(syncer.run())
     await asyncio.sleep(0)  # Yield control to give the LightChainSyncer a chance to start
 
     def finalizer():
         event_loop.run_until_complete(peer_pool.cancel())
         event_loop.run_until_complete(peer_chain.cancel())
         event_loop.run_until_complete(syncer.cancel())
+        event_loop.run_until_complete(server_request_handler.cancel())
 
     request.addfinalizer(finalizer)
 

trinity/p2p/handlers.py

@@ -92,10 +92,12 @@ async def _local_chain_info(self) -> ChainInfo:
 
 class BaseChainPeerFactory(BasePeerFactory):
     context: ChainContext
+    peer_class: Type[BaseChainPeer]
 
 
 class BaseChainPeerPool(BasePeerPool):
     connected_nodes: Dict[Node, BaseChainPeer]  # type: ignore
+    peer_factory_class: Type[BaseChainPeerFactory]
 
     @property
     def highest_td_peer(self) -> BaseChainPeer: