Merge pull request ethereum#620 from carver/bundled-cleanups

Bundled cleanups

Author: carver

p2p/service.py

@@ -265,13 +265,26 @@ async def cleanup(self) -> None:
                 for child_service in self._child_services])
             self.logger.debug("All child services finished")
         if self._tasks:
-            self.logger.debug("Waiting for tasks: %s", list(self._tasks))
+            self._log_tasks("Waiting for tasks")
             await asyncio.gather(*self._tasks)
             self.logger.debug("All tasks finished")
 
         await self._cleanup()
         self.events.cleaned_up.set()
 
+    def _log_tasks(self, message: str) -> None:
+        MAX_DISPLAY_TASKS = 50
+        task_list = list(self._tasks)
+        if len(self._tasks) > MAX_DISPLAY_TASKS:
+            task_display = ''.join(map(str, [
+                task_list[:MAX_DISPLAY_TASKS // 2],
+                '...',
+                task_list[-1 * MAX_DISPLAY_TASKS // 2:],
+            ]))
+        else:
+            task_display = str(task_list)
+        self.logger.debug("%s (%d): %s", message, len(self._tasks), task_display)
+
     def cancel_nowait(self) -> None:
         if self.is_cancelled:
             self.logger.warning("Tried to cancel %s, but it was already cancelled", self)
@@ -295,7 +308,7 @@ async def cancel(self) -> None:
                 "Timed out waiting for %s to finish its cleanup, forcibly cancelling pending "
                 "tasks and exiting anyway", self)
             if self._tasks:
-                self.logger.debug("Pending tasks: %s", list(self._tasks))
+                self._log_tasks("Pending tasks")
             if self._child_services:
                 self.logger.debug("Pending child services: %s", list(self._child_services))
             self._forcibly_cancel_all_tasks()

trinity/protocol/eth/commands.py

@@ -18,20 +18,23 @@
 from trinity.rlp.sedes import HashOrNumber
 
 
+hash_sedes = sedes.Binary(min_length=32, max_length=32)
+
+
 class Status(Command):
     _cmd_id = 0
     structure = (
         ('protocol_version', sedes.big_endian_int),
         ('network_id', sedes.big_endian_int),
         ('td', sedes.big_endian_int),
-        ('best_hash', sedes.binary),
-        ('genesis_hash', sedes.binary),
+        ('best_hash', hash_sedes),
+        ('genesis_hash', hash_sedes),
     )
 
 
 class NewBlockHashes(Command):
     _cmd_id = 1
-    structure = sedes.CountableList(sedes.List([sedes.binary, sedes.big_endian_int]))
+    structure = sedes.CountableList(sedes.List([hash_sedes, sedes.big_endian_int]))
 
 
 class Transactions(Command):
@@ -79,7 +82,7 @@ class NewBlock(Command):
 
 class GetNodeData(Command):
     _cmd_id = 13
-    structure = sedes.CountableList(sedes.binary)
+    structure = sedes.CountableList(hash_sedes)
 
 
 class NodeData(Command):
@@ -89,7 +92,7 @@ class NodeData(Command):
 
 class GetReceipts(Command):
     _cmd_id = 15
-    structure = sedes.CountableList(sedes.binary)
+    structure = sedes.CountableList(hash_sedes)
 
 
 class Receipts(Command):

trinity/protocol/eth/servers.py

@@ -17,10 +17,10 @@
     BlockIdentifier,
     Hash32,
 )
-
 from eth_utils import (
     to_hex,
 )
+from trie.exceptions import MissingTrieNode
 
 from p2p import protocol
 from p2p.peer import BasePeer
@@ -83,8 +83,18 @@ async def handle_get_block_bodies(self, peer: ETHPeer, block_hashes: Sequence[Ha
                     "%s asked for a block we don't have: %s", peer, to_hex(block_hash)
                 )
                 continue
-            transactions = await self.wait(
-                self.db.coro_get_block_transactions(header, BaseTransactionFields))
+            try:
+                transactions = await self.wait(
+                    self.db.coro_get_block_transactions(header, BaseTransactionFields))
+            except MissingTrieNode as exc:
+                self.logger.debug(
+                    "%s asked for block transactions we don't have: %s, "
+                    "due to %r",
+                    peer,
+                    to_hex(block_hash),
+                    exc,
+                )
+                continue
             uncles = await self.wait(self.db.coro_get_block_uncles(header.uncles_hash))
             bodies.append(BlockBody(transactions, uncles))
         self.logger.debug2("Replying to %s with %d block bodies", peer, len(bodies))
@@ -104,7 +114,17 @@ async def handle_get_receipts(self, peer: ETHPeer, block_hashes: Sequence[Hash32
                     "%s asked receipts for a block we don't have: %s", peer, to_hex(block_hash)
                 )
                 continue
-            block_receipts = await self.wait(self.db.coro_get_receipts(header, Receipt))
+            try:
+                block_receipts = await self.wait(self.db.coro_get_receipts(header, Receipt))
+            except MissingTrieNode as exc:
+                self.logger.debug(
+                    "%s asked for block receipts we don't have: %s, "
+                    "due to %r",
+                    peer,
+                    to_hex(block_hash),
+                    exc,
+                )
+                continue
             receipts.append(block_receipts)
         self.logger.debug2("Replying to %s with receipts for %d blocks", peer, len(receipts))
         peer.sub_proto.send_receipts(receipts)

trinity/sync/full/chain.py

@@ -11,6 +11,7 @@
     Awaitable,
     Callable,
     Dict,
+    Iterable,
     List,
     NamedTuple,
     FrozenSet,
@@ -24,6 +25,7 @@
 from eth_utils import (
     humanize_hash,
     humanize_seconds,
+    to_tuple,
     ValidationError,
 )
 from eth_utils.toolz import (
@@ -39,6 +41,7 @@
     EMPTY_UNCLE_HASH,
 )
 from eth.exceptions import HeaderNotFound
+from eth.rlp.blocks import BaseBlock
 from eth.rlp.headers import BlockHeader
 from eth.rlp.receipts import Receipt
 from eth.rlp.transactions import BaseTransaction
@@ -963,12 +966,16 @@ def __init__(self,
         )
         self._block_importer = block_importer
 
+        # Track if any headers have been received yet
+        self._got_first_header = asyncio.Event()
+
     async def _run(self) -> None:
         head = await self.wait(self.db.coro_get_canonical_head())
         self._block_import_tracker.set_finished_dependency(head)
         self.run_daemon_task(self._launch_prerequisite_tasks())
         self.run_daemon_task(self._assign_body_download_to_peers())
         self.run_daemon_task(self._import_ready_blocks())
+        self.run_daemon_task(self._display_stats())
         await super()._run()
 
     def register_peer(self, peer: BasePeer) -> None:
@@ -1045,21 +1052,9 @@ async def _import_blocks(self, headers: Tuple[BlockHeader, ...]) -> None:
 
         :param headers: headers that have the block bodies downloaded
         """
-        for header in headers:
-            vm_class = self.chain.get_vm_class(header)
-            block_class = vm_class.get_block_class()
+        unimported_blocks = self._headers_to_blocks(headers)
 
-            if _is_body_empty(header):
-                transactions: List[BaseTransaction] = []
-                uncles: List[BlockHeader] = []
-            else:
-                body = self._pending_bodies.pop(header)
-                tx_class = block_class.get_transaction_class()
-                transactions = [tx_class.from_base_transaction(tx)
-                                for tx in body.transactions]
-                uncles = body.uncles
-
-            block = block_class(header, transactions, uncles)
+        for block in unimported_blocks:
             timer = Timer()
             _, new_canonical_blocks, old_canonical_blocks = await self.wait(
                 self._block_importer.import_block(block)
@@ -1068,24 +1063,57 @@ async def _import_blocks(self, headers: Tuple[BlockHeader, ...]) -> None:
             if new_canonical_blocks == (block,):
                 # simple import of a single new block.
                 self.logger.info("Imported block %d (%d txs) in %.2f seconds",
-                                 block.number, len(transactions), timer.elapsed)
+                                 block.number, len(block.transactions), timer.elapsed)
             elif not new_canonical_blocks:
                 # imported block from a fork.
                 self.logger.info("Imported non-canonical block %d (%d txs) in %.2f seconds",
-                                 block.number, len(transactions), timer.elapsed)
+                                 block.number, len(block.transactions), timer.elapsed)
             elif old_canonical_blocks:
                 self.logger.info(
                     "Chain Reorganization: Imported block %d (%d txs) in %.2f "
                     "seconds, %d blocks discarded and %d new canonical blocks added",
                     block.number,
-                    len(transactions),
+                    len(block.transactions),
                     timer.elapsed,
                     len(old_canonical_blocks),
                     len(new_canonical_blocks),
                 )
             else:
                 raise Exception("Invariant: unreachable code path")
 
+    @to_tuple
+    def _headers_to_blocks(self, headers: Iterable[BlockHeader]) -> Iterable[BaseBlock]:
+        for header in headers:
+            vm_class = self.chain.get_vm_class(header)
+            block_class = vm_class.get_block_class()
+
+            if _is_body_empty(header):
+                transactions: List[BaseTransaction] = []
+                uncles: List[BlockHeader] = []
+            else:
+                body = self._pending_bodies.pop(header)
+                tx_class = block_class.get_transaction_class()
+                transactions = [tx_class.from_base_transaction(tx)
+                                for tx in body.transactions]
+                uncles = body.uncles
+
+            yield block_class(header, transactions, uncles)
+
+    async def _display_stats(self) -> None:
+        self.logger.debug("Regular sync waiting for first header to arrive")
+        await self.wait(self._got_first_header.wait())
+        self.logger.debug("Regular sync first header arrived")
+
+        while self.is_operational:
+            await self.sleep(5)
+            self.logger.debug(
+                "(in progress, queued, max size) of bodies, receipts: %r. Write capacity? %s",
+                [(q.num_in_progress(), len(q), q._maxsize) for q in (
+                    self._block_body_tasks,
+                )],
+                "yes" if self._db_buffer_capacity.is_set() else "no",
+            )
+
 
 def _is_body_empty(header: BlockHeader) -> bool:
     return header.transaction_root == BLANK_ROOT_HASH and header.uncles_hash == EMPTY_UNCLE_HASH

trinity/sync/full/state.py

@@ -86,7 +86,7 @@ def __init__(self,
         self.root_hash = root_hash
         # We use a LevelDB instance for the nodes cache because a full state download, if run
         # uninterrupted will visit more than 180M nodes, making an in-memory cache unfeasible.
-        self._nodes_cache_dir = tempfile.TemporaryDirectory(prefix="pyevm-state-sync-cache")
+        self._nodes_cache_dir = tempfile.TemporaryDirectory(prefix="trinity-state-sync-cache")
 
         # Allow the LevelDB instance to consume half of the entire file descriptor limit that
         # the OS permits. Let the other half be reserved for other db access, networking etc.