StateDownloader now handles block body/receipt and trie requests

Turned the ChainSyncer methods that do that into standalone funcs
so they could be reused in StateDownloader

Author: gsalgado

p2p/chain.py

@@ -564,55 +564,17 @@ async def _handle_msg(self, peer: HeaderRequestingPeer, cmd: protocol.Command,
         elif isinstance(cmd, eth.GetBlockHeaders):
             await self._handle_get_block_headers(peer, cast(Dict[str, Any], msg))
         elif isinstance(cmd, eth.GetBlockBodies):
-            await self._handle_get_block_bodies(peer, cast(List[Hash32], msg))
+            await handle_get_block_bodies(
+                self.db, peer, cast(List[Hash32], msg), self.logger, self.cancel_token)
         elif isinstance(cmd, eth.GetReceipts):
-            await self._handle_get_receipts(peer, cast(List[Hash32], msg))
+            await handle_get_receipts(
+                self.db, peer, cast(List[Hash32], msg), self.logger, self.cancel_token)
         elif isinstance(cmd, eth.GetNodeData):
-            await self._handle_get_node_data(peer, cast(List[Hash32], msg))
+            await handle_get_node_data(
+                self.db, peer, cast(List[Hash32], msg), self.logger, self.cancel_token)
         else:
             self.logger.debug("%s msg not handled yet, need to be implemented", cmd)
 
-    async def _handle_get_block_bodies(self, peer: ETHPeer, msg: List[Hash32]) -> None:
-        bodies = []
-        # Only serve up to eth.MAX_BODIES_FETCH items in every request.
-        hashes = msg[:eth.MAX_BODIES_FETCH]
-        for block_hash in hashes:
-            try:
-                header = await self.wait(self.db.coro_get_block_header_by_hash(block_hash))
-            except HeaderNotFound:
-                self.logger.debug("%s asked for block we don't have: %s", peer, block_hash)
-                continue
-            transactions = await self.wait(
-                self.db.coro_get_block_transactions(header, BaseTransactionFields))
-            uncles = await self.wait(self.db.coro_get_block_uncles(header.uncles_hash))
-            bodies.append(BlockBody(transactions, uncles))
-        peer.sub_proto.send_block_bodies(bodies)
-
-    async def _handle_get_receipts(self, peer: ETHPeer, msg: List[Hash32]) -> None:
-        receipts = []
-        # Only serve up to eth.MAX_RECEIPTS_FETCH items in every request.
-        hashes = msg[:eth.MAX_RECEIPTS_FETCH]
-        for block_hash in hashes:
-            try:
-                header = await self.wait(self.db.coro_get_block_header_by_hash(block_hash))
-            except HeaderNotFound:
-                self.logger.debug(
-                    "%s asked receipts for block we don't have: %s", peer, block_hash)
-                continue
-            receipts.append(await self.wait(self.db.coro_get_receipts(header, Receipt)))
-        peer.sub_proto.send_receipts(receipts)
-
-    async def _handle_get_node_data(self, peer: ETHPeer, msg: List[Hash32]) -> None:
-        nodes = []
-        for node_hash in msg:
-            try:
-                node = await self.db.coro_get(node_hash)
-            except KeyError:
-                self.logger.debug("%s asked for a trie node we don't have: %s", peer, node_hash)
-                continue
-            nodes.append(node)
-        peer.sub_proto.send_node_data(nodes)
-
     async def _process_headers(
             self, peer: HeaderRequestingPeer, headers: Tuple[BlockHeader, ...]) -> int:
         target_td = await self._calculate_td(headers)
@@ -756,6 +718,67 @@ async def lookup_headers(
     return headers
 
 
+async def handle_get_block_bodies(
+        chaindb: 'AsyncChainDB', peer: ETHPeer, block_hashes: List[Hash32],
+        logger: logging.Logger, token: CancelToken) -> None:
+    bodies = []
+    # Only serve up to eth.MAX_BODIES_FETCH items in every request.
+    for block_hash in block_hashes[:eth.MAX_BODIES_FETCH]:
+        try:
+            header = await wait_with_token(
+                chaindb.coro_get_block_header_by_hash(block_hash),
+                token=token)
+        except HeaderNotFound:
+            logger.debug("%s asked for block we don't have: %s", peer, block_hash)
+            continue
+        transactions = await wait_with_token(
+            chaindb.coro_get_block_transactions(header, BaseTransactionFields),
+            token=token)
+        uncles = await wait_with_token(
+            chaindb.coro_get_block_uncles(header.uncles_hash),
+            token=token)
+        bodies.append(BlockBody(transactions, uncles))
+    peer.sub_proto.send_block_bodies(bodies)
+
+
+async def handle_get_receipts(
+        chaindb: 'AsyncChainDB', peer: ETHPeer, block_hashes: List[Hash32],
+        logger: logging.Logger, token: CancelToken) -> None:
+    receipts = []
+    # Only serve up to eth.MAX_RECEIPTS_FETCH items in every request.
+    for block_hash in block_hashes[:eth.MAX_RECEIPTS_FETCH]:
+        try:
+            header = await wait_with_token(
+                chaindb.coro_get_block_header_by_hash(block_hash),
+                token=token)
+        except HeaderNotFound:
+            logger.debug(
+                "%s asked receipts for block we don't have: %s", peer, block_hash)
+            continue
+        block_receipts = await wait_with_token(
+            chaindb.coro_get_receipts(header, Receipt),
+            token=token)
+        receipts.append(block_receipts)
+    peer.sub_proto.send_receipts(receipts)
+
+
+async def handle_get_node_data(
+        chaindb: 'AsyncChainDB', peer: ETHPeer, node_hashes: List[Hash32],
+        logger: logging.Logger, token: CancelToken) -> None:
+    nodes = []
+    # Only serve up to eth.MAX_STATE_FETCH items in every request.
+    for node_hash in node_hashes[:eth.MAX_STATE_FETCH]:
+        try:
+            node = await wait_with_token(
+                chaindb.coro_get(node_hash),
+                token=token)
+        except KeyError:
+            logger.debug("%s asked for a trie node we don't have: %s", peer, node_hash)
+            continue
+        nodes.append(node)
+    peer.sub_proto.send_node_data(nodes)
+
+
 def _test() -> None:
     import argparse
     import signal

p2p/state.py

@@ -39,7 +39,8 @@
 
 from p2p import eth
 from p2p import protocol
-from p2p.chain import lookup_headers
+from p2p.chain import (
+    handle_get_block_bodies, handle_get_node_data, handle_get_receipts, lookup_headers)
 from p2p.cancel_token import CancelToken
 from p2p.exceptions import OperationCancelled
 from p2p.peer import ETHPeer, PeerPool, PeerPoolSubscriber
@@ -127,6 +128,15 @@ async def _handle_msg(
                 self._pending_nodes.pop(node_key, None)
         elif isinstance(cmd, eth.GetBlockHeaders):
             await self._handle_get_block_headers(peer, cast(Dict[str, Any], msg))
+        elif isinstance(cmd, eth.GetBlockBodies):
+            await handle_get_block_bodies(
+                self.chaindb, peer, cast(List[Hash32], msg), self.logger, self.cancel_token)
+        elif isinstance(cmd, eth.GetReceipts):
+            await handle_get_receipts(
+                self.chaindb, peer, cast(List[Hash32], msg), self.logger, self.cancel_token)
+        elif isinstance(cmd, eth.GetNodeData):
+            await handle_get_node_data(
+                self.chaindb, peer, cast(List[Hash32], msg), self.logger, self.cancel_token)
         else:
             self.logger.warn("%s not handled during StateSync, must be implemented", cmd)
 
@@ -208,7 +218,7 @@ async def _run(self) -> None:
                     # received nodes (scheduling new requests), there may be cases when the
                     # pending nodes take a while to arrive thus causing the scheduler to run out
                     # of new requests for a while.
-                    self.logger.info("Scheduler queue is empty, sleeping a bit")
+                    self.logger.debug("Scheduler queue is empty, sleeping a bit")
                     await self.wait_first(asyncio.sleep(0.5))
                     continue
 
@@ -298,7 +308,7 @@ async def run() -> None:
         downloader.logger.info("run() finished, exiting")
         sigint_received.set()
 
-    loop.set_debug(True)
+    # loop.set_debug(True)
     asyncio.ensure_future(exit_on_sigint())
     asyncio.ensure_future(run())
     loop.run_forever()