Move discovery into an isolated plugin

Resolves #1518

Author: cburgdorf

p2p/constants.py

@@ -76,3 +76,8 @@
 # The amount of time that the BasePeerPool will wait for a peer to boot before
 # aborting the connection attempt.
 DEFAULT_PEER_BOOT_TIMEOUT = 20
+
+# Interval at which peer pool is checked for potential new candidates
+DISOVERY_INTERVAL = 2
+# Timeout used when fetching peer candidates from discovery
+REQUEST_PEER_CANDIDATE_TIMEOUT = 0.5

p2p/discovery.py

@@ -32,6 +32,10 @@
 
 import cytoolz
 
+from lahja import (
+    Endpoint,
+)
+
 import rlp
 
 from eth_typing import Hash32
@@ -56,10 +60,11 @@
 
 from cancel_token import CancelToken, OperationCancelled
 
+from p2p.events import PeerCandidatesRequest, RandomBootnodeRequest
 from p2p.exceptions import AlreadyWaitingDiscoveryResponse, NoEligibleNodes, UnableToGetDiscV5Ticket
+from p2p.kademlia import to_uris
 from p2p import kademlia
 from p2p import protocol
-from p2p.peer import BasePeerPool
 from p2p.service import BaseService
 
 if TYPE_CHECKING:
@@ -951,21 +956,48 @@ class DiscoveryService(BaseService):
     _last_lookup: float = 0
     _lookup_interval: int = 30
 
-    def __init__(self, proto: DiscoveryProtocol, peer_pool: BasePeerPool,
-                 port: int, token: CancelToken = None) -> None:
+    def __init__(self,
+                 proto: DiscoveryProtocol,
+                 port: int,
+                 event_bus: Endpoint,
+                 token: CancelToken = None) -> None:
         super().__init__(token)
         self.proto = proto
-        self.peer_pool = peer_pool
         self.port = port
+        self._event_bus = event_bus
         self._lookup_running = asyncio.Lock()
 
+        self.run_daemon_task(self.handle_get_peer_candidates_requests())
+        self.run_daemon_task(self.handle_get_random_bootnode_requests())
+
+    async def handle_get_peer_candidates_requests(self) -> None:
+        async for event in self._event_bus.stream(PeerCandidatesRequest):
+
+            self.run_task(self.maybe_lookup_random_node())
+
+            nodes = tuple(to_uris(self.proto.get_nodes_to_connect(event.max_candidates)))
+
+            self.logger.debug2("Broadcasting peer candidates (%s)", nodes)
+            self._event_bus.broadcast(
+                event.expected_response_type()(nodes),
+                event.broadcast_config()
+            )
+
+    async def handle_get_random_bootnode_requests(self) -> None:
+        async for event in self._event_bus.stream(RandomBootnodeRequest):
+
+            nodes = tuple(to_uris(self.proto.get_random_bootnode()))
+
+            self.logger.debug2("Broadcasting random boot nodes (%s)", nodes)
+            self._event_bus.broadcast(
+                event.expected_response_type()(nodes),
+                event.broadcast_config()
+            )
+
     async def _run(self) -> None:
         await self._start_udp_listener()
-        connect_loop_sleep = 2
         self.run_task(self.proto.bootstrap())
-        while self.is_operational:
-            await self.maybe_connect_to_more_peers()
-            await self.sleep(connect_loop_sleep)
+        await self.cancel_token.wait()
 
     async def _start_udp_listener(self) -> None:
         loop = asyncio.get_event_loop()
@@ -975,22 +1007,6 @@ async def _start_udp_listener(self) -> None:
             local_addr=('0.0.0.0', self.port),
             family=socket.AF_INET)
 
-    async def maybe_connect_to_more_peers(self) -> None:
-        """Connect to more peers if we're not yet maxed out to max_peers"""
-        if self.peer_pool.is_full:
-            self.logger.debug("Already connected to %s peers; sleeping", len(self.peer_pool))
-            return
-
-        self.run_task(self.maybe_lookup_random_node())
-
-        await self.peer_pool.connect_to_nodes(
-            self.proto.get_nodes_to_connect(self.peer_pool.max_peers))
-
-        # In some cases (e.g ROPSTEN or private testnets), the discovery table might be full of
-        # bad peers so if we can't connect to any peers we try a random bootstrap node as well.
-        if not len(self.peer_pool):
-            await self.peer_pool.connect_to_nodes(self.proto.get_random_bootnode())
-
     async def maybe_lookup_random_node(self) -> None:
         if self._last_lookup + self._lookup_interval > time.time():
             return

p2p/events.py

@@ -1,4 +1,5 @@
 from typing import (
+    Tuple,
     Type,
 )
 
@@ -8,6 +9,36 @@
 )
 
 
+class BaseDiscoveryServiceResponse(BaseEvent):
+
+    def __init__(self, error: Exception) -> None:
+        self.error = error
+
+
+class PeerCandidatesResponse(BaseDiscoveryServiceResponse):
+
+    def __init__(self, candidates: Tuple[str, ...], error: Exception=None) -> None:
+        super().__init__(error)
+        self.candidates = candidates
+
+
+class PeerCandidatesRequest(BaseRequestResponseEvent[PeerCandidatesResponse]):
+
+    def __init__(self, max_candidates: int) -> None:
+        self.max_candidates = max_candidates
+
+    @staticmethod
+    def expected_response_type() -> Type[PeerCandidatesResponse]:
+        return PeerCandidatesResponse
+
+
+class RandomBootnodeRequest(BaseRequestResponseEvent[PeerCandidatesResponse]):
+
+    @staticmethod
+    def expected_response_type() -> Type[PeerCandidatesResponse]:
+        return PeerCandidatesResponse
+
+
 class PeerCountResponse(BaseEvent):
 
     def __init__(self, peer_count: int) -> None:

p2p/kademlia.py

@@ -105,6 +105,9 @@ def from_uri(cls, uri: str) -> 'Node':
         pubkey = keys.PublicKey(decode_hex(parsed.username))
         return cls(pubkey, Address(parsed.hostname, parsed.port))
 
+    def uri(self) -> str:
+        return f'enode://{self.pubkey.to_hex()}@{self.address.ip}:{self.address.tcp_port}'
+
     def __str__(self) -> str:
         return '<Node(%s@%s)>' % (self.pubkey.to_hex()[:6], self.address.ip)
 
@@ -133,6 +136,19 @@ def __ne__(self, other: object) -> bool:
     def __hash__(self) -> int:
         return hash(self.pubkey)
 
+# TODO: check if we can make the nodes pickable and get rid of these
+# https://github.com/ethereum/py-evm/issues/1578
+
+
+def to_uris(nodes: Iterable[Node]) -> Iterator[str]:
+    for node in nodes:
+        yield node.uri()
+
+
+def from_uris(uris: Iterable[str]) -> Iterator[Node]:
+    for uri in uris:
+        yield Node.from_uri(uri)
+
 
 @total_ordering
 class KBucket(Sized):

p2p/peer.py

@@ -48,7 +48,10 @@
 
 from p2p import auth
 from p2p import protocol
-from p2p.kademlia import Node
+from p2p.kademlia import (
+    from_uris,
+    Node,
+)
 from p2p.exceptions import (
     BadAckMessage,
     DecryptionError,
@@ -81,13 +84,17 @@
     CONN_IDLE_TIMEOUT,
     DEFAULT_MAX_PEERS,
     DEFAULT_PEER_BOOT_TIMEOUT,
+    DISOVERY_INTERVAL,
     HEADER_LEN,
     MAC_LEN,
+    REQUEST_PEER_CANDIDATE_TIMEOUT,
 )
 
 from .events import (
+    PeerCandidatesRequest,
     PeerCountRequest,
     PeerCountResponse,
+    RandomBootnodeRequest,
 )
 
 
@@ -799,6 +806,43 @@ async def handle_peer_count_requests(self) -> None:
                 # `event.broadcast_config()` API.
                 self.event_bus.broadcast(PeerCountResponse(len(self)), req.broadcast_config())
 
+    async def maybe_connect_more_peers(self) -> None:
+        while self.is_operational:
+            await self.sleep(DISOVERY_INTERVAL)
+
+            available_peer_slots = self.max_peers - len(self)
+            if available_peer_slots > 0:
+                try:
+                    response = await self.wait(
+                        # TODO: This should use a BroadcastConfig to send the request to discovery
+                        # only as soon as we have cut a new Lahja release.
+                        self.event_bus.request(PeerCandidatesRequest(available_peer_slots)),
+                        timeout=REQUEST_PEER_CANDIDATE_TIMEOUT
+                    )
+                except TimeoutError:
+                    self.logger.warning("Discovery did not answer PeerCandidateRequest in time")
+                    continue
+
+                # In some cases (e.g ROPSTEN or private testnets), the discovery table might be
+                # full of bad peers so if we can't connect to any peers we try a random bootstrap
+                # node as well.
+                if not len(self):
+                    try:
+                        response = await self.wait(
+                            # TODO: This should use a BroadcastConfig to send the request to
+                            # discovery only as soon as we have cut a new Lahja release.
+                            self.event_bus.request(RandomBootnodeRequest()),
+                            timeout=REQUEST_PEER_CANDIDATE_TIMEOUT
+                        )
+                    except TimeoutError:
+                        self.logger.warning(
+                            "Discovery did not answer RandomBootnodeRequest in time"
+                        )
+                        continue
+
+                self.logger.debug2("Received candidates to connect to (%s)", response.candidates)
+                await self.connect_to_nodes(from_uris(response.candidates))
+
     def __len__(self) -> int:
         return len(self.connected_nodes)
 
@@ -880,6 +924,7 @@ async def _run(self) -> None:
         # so in order to ensure we cancel all peers when we terminate.
         if self.event_bus is not None:
             self.run_daemon_task(self.handle_peer_count_requests())
+            self.run_daemon_task(self.maybe_connect_more_peers())
         self.run_daemon_task(self._periodically_report_stats())
         await self.cancel_token.wait()
 

trinity/nodes/base.py

@@ -52,7 +52,6 @@ def __init__(self, event_bus: Endpoint, trinity_config: TrinityConfig) -> None:
 
         self._jsonrpc_ipc_path: Path = trinity_config.jsonrpc_ipc_path
         self._network_id = trinity_config.network_id
-        self._use_discv5 = trinity_config.use_discv5
 
         self.event_bus = event_bus
 

trinity/nodes/full.py

@@ -44,7 +44,6 @@ def get_p2p_server(self) -> FullServer:
                 max_peers=self._max_peers,
                 bootstrap_nodes=self._bootstrap_nodes,
                 preferred_nodes=self._preferred_nodes,
-                use_discv5=self._use_discv5,
                 token=self.cancel_token,
                 event_bus=self.event_bus,
             )

trinity/nodes/light.py

@@ -80,7 +80,6 @@ def get_p2p_server(self) -> LightServer:
                 max_peers=self._max_peers,
                 bootstrap_nodes=self._bootstrap_nodes,
                 preferred_nodes=self._preferred_nodes,
-                use_discv5=self._use_discv5,
                 token=self.cancel_token,
                 event_bus=self.event_bus,
             )