Merge pull request #1120 from pipermerriam/piper/start-extraction-of-trinity-from-p2p-module

WIP: Start moving the ETH p2p protocol stuff into the trinity module

Author: pipermerriam

p2p/chain.py

@@ -39,13 +39,11 @@
 from eth.utils.logging import TraceLogger
 
 from p2p import protocol
-from p2p import eth
-from p2p import les
 from p2p.cancellable import CancellableMixin
 from p2p.constants import MAX_REORG_DEPTH, SEAL_CHECK_RANDOM_SAMPLE_RATE
 from p2p.exceptions import NoEligiblePeers, ValidationError
 from p2p.p2p_proto import DisconnectReason
-from p2p.peer import BasePeer, ETHPeer, LESPeer, PeerPool, PeerSubscriber
+from p2p.peer import BasePeer, PeerPool, PeerSubscriber
 from p2p.rlp import BlockBody
 from p2p.service import BaseService
 from p2p.utils import (
@@ -57,9 +55,12 @@
 if TYPE_CHECKING:
     from trinity.db.chain import AsyncChainDB  # noqa: F401
     from trinity.db.header import AsyncHeaderDB  # noqa: F401
+    from trinity.protocol.eth.peer import ETHPeer  # noqa: F401
+    from trinity.protocol.les.peer import LESPeer  # noqa: F401
+    from trinity.protocol.base_request import BaseHeaderRequest  # noqa: F401
 
 
-HeaderRequestingPeer = Union[LESPeer, ETHPeer]
+HeaderRequestingPeer = Union['LESPeer', 'ETHPeer']
 
 
 class BaseHeaderChainSyncer(BaseService, PeerSubscriber):
@@ -300,20 +301,23 @@ class LightChainSyncer(BaseHeaderChainSyncer):
 
     async def _handle_msg(self, peer: HeaderRequestingPeer, cmd: protocol.Command,
                           msg: protocol._DecodedMsgType) -> None:
-        if isinstance(cmd, les.Announce):
+        from trinity.protocol.les import commands
+        from trinity.protocol.les.peer import LESPeer  # noqa: F811
+        if isinstance(cmd, commands.Announce):
             self._sync_requests.put_nowait(peer)
-        elif isinstance(cmd, les.GetBlockHeaders):
+        elif isinstance(cmd, commands.GetBlockHeaders):
             msg = cast(Dict[str, Any], msg)
             await self._handle_get_block_headers(cast(LESPeer, peer), msg)
-        elif isinstance(cmd, les.BlockHeaders):
+        elif isinstance(cmd, commands.BlockHeaders):
             # `BlockHeaders` messages are handled at the peer level.
             pass
         else:
             self.logger.debug("Ignoring %s message from %s", cmd, peer)
 
-    async def _handle_get_block_headers(self, peer: LESPeer, msg: Dict[str, Any]) -> None:
+    async def _handle_get_block_headers(self, peer: 'LESPeer', msg: Dict[str, Any]) -> None:
+        from trinity.protocol.les.requests import HeaderRequest
         self.logger.debug("Peer %s made header request: %s", peer, msg)
-        request = les.HeaderRequest(
+        request = HeaderRequest(
             msg['query'].block_number_or_hash,
             msg['query'].max_headers,
             msg['query'].skip,
@@ -356,8 +360,8 @@ def __init__(self,
         super().__init__(chain, db, peer_pool, token)
         # Those are used by our msg handlers and _download_block_parts() in order to track missing
         # bodies/receipts for a given chain segment.
-        self._downloaded_receipts: asyncio.Queue[Tuple[ETHPeer, List[DownloadedBlockPart]]] = asyncio.Queue()  # noqa: E501
-        self._downloaded_bodies: asyncio.Queue[Tuple[ETHPeer, List[DownloadedBlockPart]]] = asyncio.Queue()  # noqa: E501
+        self._downloaded_receipts: asyncio.Queue[Tuple['ETHPeer', List[DownloadedBlockPart]]] = asyncio.Queue()  # noqa: E501
+        self._downloaded_bodies: asyncio.Queue[Tuple['ETHPeer', List[DownloadedBlockPart]]] = asyncio.Queue()  # noqa: E501
 
     async def _calculate_td(self, headers: Tuple[BlockHeader, ...]) -> int:
         """Return the score (total difficulty) of the last header in the given list.
@@ -483,7 +487,8 @@ def _request_block_parts(
             self,
             target_td: int,
             headers: List[BlockHeader],
-            request_func: Callable[[ETHPeer, List[BlockHeader]], None]) -> int:
+            request_func: Callable[['ETHPeer', List[BlockHeader]], None]) -> int:
+        from trinity.protocol.eth.peer import ETHPeer  # noqa: F811
         peers = self.peer_pool.get_peers(target_td)
         if not peers:
             raise NoEligiblePeers()
@@ -493,13 +498,13 @@ def _request_block_parts(
             request_func(cast(ETHPeer, peer), batch)
         return len(batches)
 
-    def _send_get_block_bodies(self, peer: ETHPeer, headers: List[BlockHeader]) -> None:
+    def _send_get_block_bodies(self, peer: 'ETHPeer', headers: List[BlockHeader]) -> None:
         block_numbers = ", ".join(str(h.block_number) for h in headers)
         self.logger.debug(
             "Requesting %d block bodies (%s) to %s", len(headers), block_numbers, peer)
         peer.sub_proto.send_get_block_bodies([header.hash for header in headers])
 
-    def _send_get_receipts(self, peer: ETHPeer, headers: List[BlockHeader]) -> None:
+    def _send_get_receipts(self, peer: 'ETHPeer', headers: List[BlockHeader]) -> None:
         block_numbers = ", ".join(str(h.block_number) for h in headers)
         self.logger.debug(
             "Requesting %d block receipts (%s) to %s", len(headers), block_numbers, peer)
@@ -527,47 +532,54 @@ def request_receipts(self, target_td: int, headers: List[BlockHeader]) -> int:
 
     async def _handle_msg(self, peer: HeaderRequestingPeer, cmd: protocol.Command,
                           msg: protocol._DecodedMsgType) -> None:
+        from trinity.protocol.eth.peer import ETHPeer  # noqa: F811
+        from trinity.protocol.eth import commands
+        from trinity.protocol.eth import (
+            constants as eth_constants,
+        )
+
         peer = cast(ETHPeer, peer)
-        if isinstance(cmd, eth.BlockBodies):
+
+        if isinstance(cmd, commands.BlockBodies):
             await self._handle_block_bodies(peer, list(cast(Tuple[BlockBody], msg)))
-        elif isinstance(cmd, eth.Receipts):
+        elif isinstance(cmd, commands.Receipts):
             await self._handle_block_receipts(peer, cast(List[List[Receipt]], msg))
-        elif isinstance(cmd, eth.NewBlock):
+        elif isinstance(cmd, commands.NewBlock):
             await self._handle_new_block(peer, cast(Dict[str, Any], msg))
-        elif isinstance(cmd, eth.GetBlockHeaders):
+        elif isinstance(cmd, commands.GetBlockHeaders):
             await self._handle_get_block_headers(peer, cast(Dict[str, Any], msg))
-        elif isinstance(cmd, eth.BlockHeaders):
+        elif isinstance(cmd, commands.BlockHeaders):
             # `BlockHeaders` messages are handled at the peer level.
             pass
-        elif isinstance(cmd, eth.GetBlockBodies):
-            # Only serve up to eth.MAX_BODIES_FETCH items in every request.
-            block_hashes = cast(List[Hash32], msg)[:eth.MAX_BODIES_FETCH]
+        elif isinstance(cmd, commands.GetBlockBodies):
+            # Only serve up to MAX_BODIES_FETCH items in every request.
+            block_hashes = cast(List[Hash32], msg)[:eth_constants.MAX_BODIES_FETCH]
             await self._handler.handle_get_block_bodies(peer, block_hashes)
-        elif isinstance(cmd, eth.GetReceipts):
-            # Only serve up to eth.MAX_RECEIPTS_FETCH items in every request.
-            block_hashes = cast(List[Hash32], msg)[:eth.MAX_RECEIPTS_FETCH]
+        elif isinstance(cmd, commands.GetReceipts):
+            # Only serve up to MAX_RECEIPTS_FETCH items in every request.
+            block_hashes = cast(List[Hash32], msg)[:eth_constants.MAX_RECEIPTS_FETCH]
             await self._handler.handle_get_receipts(peer, block_hashes)
-        elif isinstance(cmd, eth.GetNodeData):
-            # Only serve up to eth.MAX_STATE_FETCH items in every request.
-            node_hashes = cast(List[Hash32], msg)[:eth.MAX_STATE_FETCH]
+        elif isinstance(cmd, commands.GetNodeData):
+            # Only serve up to MAX_STATE_FETCH items in every request.
+            node_hashes = cast(List[Hash32], msg)[:eth_constants.MAX_STATE_FETCH]
             await self._handler.handle_get_node_data(peer, node_hashes)
-        elif isinstance(cmd, eth.Transactions):
+        elif isinstance(cmd, commands.Transactions):
             # Transactions msgs are handled by our TxPool service.
             pass
-        elif isinstance(cmd, eth.NodeData):
+        elif isinstance(cmd, commands.NodeData):
             # When doing a chain sync we never send GetNodeData requests, so peers should not send
             # us NodeData msgs.
             self.logger.warn("Unexpected NodeData msg from %s, disconnecting", peer)
             await peer.disconnect(DisconnectReason.bad_protocol)
         else:
             self.logger.debug("%s msg not handled yet, need to be implemented", cmd)
 
-    async def _handle_new_block(self, peer: ETHPeer, msg: Dict[str, Any]) -> None:
+    async def _handle_new_block(self, peer: 'ETHPeer', msg: Dict[str, Any]) -> None:
         self._sync_requests.put_nowait(peer)
 
     async def _handle_block_receipts(self,
-                                     peer: ETHPeer,
-                                     receipts_by_block: List[List[eth.Receipt]]) -> None:
+                                     peer: 'ETHPeer',
+                                     receipts_by_block: List[List[Receipt]]) -> None:
         self.logger.debug("Got Receipts for %d blocks from %s", len(receipts_by_block), peer)
         iterator = map(make_trie_root_and_nodes, receipts_by_block)
         # The map() call above is lazy (it returns an iterator! ;-), so it's only evaluated in
@@ -582,7 +594,7 @@ async def _handle_block_receipts(self,
         self._downloaded_receipts.put_nowait((peer, downloaded))
 
     async def _handle_block_bodies(self,
-                                   peer: ETHPeer,
+                                   peer: 'ETHPeer',
                                    bodies: List[BlockBody]) -> None:
         self.logger.debug("Got Bodies for %d blocks from %s", len(bodies), peer)
         iterator = map(make_trie_root_and_nodes, [body.transactions for body in bodies])
@@ -601,10 +613,12 @@ async def _handle_block_bodies(self,
 
     async def _handle_get_block_headers(
             self,
-            peer: ETHPeer,
+            peer: 'ETHPeer',
             query: Dict[str, Any]) -> None:
+        from trinity.protocol.eth.requests import HeaderRequest  # noqa: F811
+
         self.logger.debug("Peer %s made header request: %s", peer, query)
-        request = eth.HeaderRequest(
+        request = HeaderRequest(
             query['block_number_or_hash'],
             query['max_headers'],
             query['skip'],
@@ -626,7 +640,7 @@ class RegularChainSyncer(FastChainSyncer):
     _seal_check_random_sample_rate = 1
 
     async def _handle_block_receipts(
-            self, peer: ETHPeer, receipts_by_block: List[List[eth.Receipt]]) -> None:
+            self, peer: 'ETHPeer', receipts_by_block: List[List[Receipt]]) -> None:
         # When doing a regular sync we never request receipts.
         self.logger.warn("Unexpected BlockReceipts msg from %s, disconnecting", peer)
         await peer.disconnect(DisconnectReason.bad_protocol)
@@ -675,7 +689,7 @@ def __init__(self, db: 'AsyncHeaderDB', logger: TraceLogger, token: CancelToken)
         self.logger = logger
         self.cancel_token = token
 
-    async def handle_get_block_bodies(self, peer: ETHPeer, block_hashes: List[Hash32]) -> None:
+    async def handle_get_block_bodies(self, peer: 'ETHPeer', block_hashes: List[Hash32]) -> None:
         self.logger.trace("%s requested bodies for %d blocks", peer, len(block_hashes))
         chaindb = cast('AsyncChainDB', self.db)
         bodies = []
@@ -692,7 +706,7 @@ async def handle_get_block_bodies(self, peer: ETHPeer, block_hashes: List[Hash32
         self.logger.trace("Replying to %s with %d block bodies", peer, len(bodies))
         peer.sub_proto.send_block_bodies(bodies)
 
-    async def handle_get_receipts(self, peer: ETHPeer, block_hashes: List[Hash32]) -> None:
+    async def handle_get_receipts(self, peer: 'ETHPeer', block_hashes: List[Hash32]) -> None:
         self.logger.trace("%s requested receipts for %d blocks", peer, len(block_hashes))
         chaindb = cast('AsyncChainDB', self.db)
         receipts = []
@@ -708,7 +722,7 @@ async def handle_get_receipts(self, peer: ETHPeer, block_hashes: List[Hash32]) -
         self.logger.trace("Replying to %s with receipts for %d blocks", peer, len(receipts))
         peer.sub_proto.send_receipts(receipts)
 
-    async def handle_get_node_data(self, peer: ETHPeer, node_hashes: List[Hash32]) -> None:
+    async def handle_get_node_data(self, peer: 'ETHPeer', node_hashes: List[Hash32]) -> None:
         self.logger.trace("%s requested %d trie nodes", peer, len(node_hashes))
         chaindb = cast('AsyncChainDB', self.db)
         nodes = []
@@ -723,7 +737,7 @@ async def handle_get_node_data(self, peer: ETHPeer, node_hashes: List[Hash32]) -
         peer.sub_proto.send_node_data(nodes)
 
     async def lookup_headers(self,
-                             request: protocol.BaseHeaderRequest) -> Tuple[BlockHeader, ...]:
+                             request: 'BaseHeaderRequest') -> Tuple[BlockHeader, ...]:
         """
         Lookup :max_headers: headers starting at :block_number_or_hash:, skipping :skip: items
         between each, in reverse order if :reverse: is True.
@@ -744,7 +758,7 @@ async def lookup_headers(self,
         return headers
 
     async def _get_block_numbers_for_request(self,
-                                             request: protocol.BaseHeaderRequest
+                                             request: 'BaseHeaderRequest'
                                              ) -> Tuple[BlockNumber, ...]:
         """
         Generate the block numbers for a given `HeaderRequest`.
@@ -823,6 +837,8 @@ def _test() -> None:
     from tests.p2p.integration_test_helpers import (
         FakeAsyncChainDB, FakeAsyncMainnetChain, FakeAsyncRopstenChain, FakeAsyncHeaderDB,
         connect_to_peers_loop)
+    from trinity.protocol.eth.peer import ETHPeer  # noqa: F811
+    from trinity.protocol.les.peer import LESPeer  # noqa: F811
     from trinity.utils.chains import load_nodekey
 
     parser = argparse.ArgumentParser()

p2p/constants.py

@@ -44,10 +44,6 @@
 # Default timeout before giving up on a caller-initiated interaction
 COMPLETION_TIMEOUT = 5
 
-# Types of LES Announce messages
-LES_ANNOUNCE_SIMPLE = 1
-LES_ANNOUNCE_SIGNED = 2
-
 MAINNET_BOOTNODES = (
     'enode://a979fb575495b8d6db44f750317d0f4622bf4c2aa3365d6af7c284339968eef29b69ad0dce72a4d8db5ebb4968de0e3bec910127f134779fbcb0cb6d3331163c@52.16.188.185:30303',  # noqa: E501
     'enode://aa36fdf33dd030378a0168efe6ed7d5cc587fafa3cdd375854fe735a2e11ea3650ba29644e2db48368c46e1f60e716300ba49396cd63778bf8a818c09bded46f@13.93.211.84:30303',  # noqa: E501

p2p/lightchain.py

@@ -55,7 +55,6 @@
     DisconnectReason,
 )
 from p2p.peer import (
-    LESPeer,
     PeerPool,
     PeerSubscriber,
 )
@@ -64,10 +63,10 @@
     BaseService,
     service_timeout,
 )
-from p2p.utils import gen_request_id
 
 if TYPE_CHECKING:
     from trinity.db.header import BaseAsyncHeaderDB  # noqa: F401
+    from trinity.protocol.les.peer import LESPeer  # noqa: F401
 
 
 class LightPeerChain(PeerSubscriber, BaseService):
@@ -142,6 +141,9 @@ async def get_block_header_by_hash(self, block_hash: Hash32) -> BlockHeader:
     @alru_cache(maxsize=1024, cache_exceptions=False)
     @service_timeout(COMPLETION_TIMEOUT)
     async def get_block_body_by_hash(self, block_hash: Hash32) -> BlockBody:
+        from trinity.protocol.les.peer import LESPeer  # noqa: F811
+        from trinity.protocol.les.utils import gen_request_id
+
         peer = cast(LESPeer, self.peer_pool.highest_td_peer)
         self.logger.debug("Fetching block %s from %s", encode_hex(block_hash), peer)
         request_id = gen_request_id()
@@ -156,6 +158,9 @@ async def get_block_body_by_hash(self, block_hash: Hash32) -> BlockBody:
     @alru_cache(maxsize=1024, cache_exceptions=False)
     @service_timeout(COMPLETION_TIMEOUT)
     async def get_receipts(self, block_hash: Hash32) -> List[Receipt]:
+        from trinity.protocol.les.peer import LESPeer  # noqa: F811
+        from trinity.protocol.les.utils import gen_request_id
+
         peer = cast(LESPeer, self.peer_pool.highest_td_peer)
         self.logger.debug("Fetching %s receipts from %s", encode_hex(block_hash), peer)
         request_id = gen_request_id()
@@ -179,7 +184,7 @@ async def _get_account_from_peer(
             self,
             block_hash: Hash32,
             address: Address,
-            peer: LESPeer) -> Account:
+            peer: 'LESPeer') -> Account:
         key = keccak(address)
         proof = await self._get_proof(peer, block_hash, account_key=b'', key=key)
         header = await self._get_block_header_by_hash(block_hash, peer)
@@ -223,13 +228,15 @@ async def _get_contract_code_from_peer(
             block_hash: Hash32,
             address: Address,
             code_hash: Hash32,
-            peer: LESPeer) -> bytes:
+            peer: 'LESPeer') -> bytes:
         """
         A single attempt to get the contract code from the given peer
 
         :raise BadLESResponse: if the peer replies with contract code that does not match the
             account's code hash
         """
+        from trinity.protocol.les.utils import gen_request_id
+
         # request contract code
         request_id = gen_request_id()
         peer.sub_proto.send_get_contract_code(block_hash, keccak(address), request_id)
@@ -261,7 +268,7 @@ async def _raise_for_empty_code(
             block_hash: Hash32,
             address: Address,
             code_hash: Hash32,
-            peer: LESPeer) -> None:
+            peer: 'LESPeer') -> None:
         """
         A peer might return b'' if it doesn't have the block at the requested header,
         or it might maliciously return b'' when the code is non-empty. This method tries to tell the
@@ -315,12 +322,14 @@ async def _raise_for_empty_code(
                 )
             )
 
-    async def _get_block_header_by_hash(self, block_hash: Hash32, peer: LESPeer) -> BlockHeader:
+    async def _get_block_header_by_hash(self, block_hash: Hash32, peer: 'LESPeer') -> BlockHeader:
         """
         A single attempt to get the block header from the given peer.
 
         :raise BadLESResponse: if the peer replies with a header that has a different hash
         """
+        from trinity.protocol.les.utils import gen_request_id
+
         self.logger.debug("Fetching header %s from %s", encode_hex(block_hash), peer)
         request_id = gen_request_id()
         max_headers = 1
@@ -336,17 +345,19 @@ async def _get_block_header_by_hash(self, block_hash: Hash32, peer: LESPeer) ->
         return header
 
     async def _get_proof(self,
-                         peer: LESPeer,
+                         peer: 'LESPeer',
                          block_hash: bytes,
                          account_key: bytes,
                          key: bytes,
                          from_level: int = 0) -> List[bytes]:
+        from trinity.protocol.les.utils import gen_request_id
+
         request_id = gen_request_id()
         peer.sub_proto.send_get_proof(block_hash, account_key, key, from_level, request_id)
         reply = await self._wait_for_reply(request_id)
         return reply['proof']
 
-    async def _retry_on_bad_response(self, make_request_to_peer: Callable[[LESPeer], Any]) -> Any:
+    async def _retry_on_bad_response(self, make_request_to_peer: Callable[['LESPeer'], Any]) -> Any:
         """
         Make a call to a peer. If it behaves badly, drop it and retry with a different peer.
 
@@ -355,6 +366,8 @@ async def _retry_on_bad_response(self, make_request_to_peer: Callable[[LESPeer],
         :raise NoEligiblePeers: if no peers are available to fulfill the request
         :raise TimeoutError: if an individual request or the overall process times out
         """
+        from trinity.protocol.les.peer import LESPeer  # noqa: F811
+
         for _ in range(MAX_REQUEST_ATTEMPTS):
             try:
                 peer = cast(LESPeer, self.peer_pool.highest_td_peer)