Merge pull request ethereum#619 from carver/header-skipping-unification

Header skipping unification

Author: carver

trinity/_utils/headers.py

@@ -75,13 +75,28 @@ async def _skip_complete_headers_iterator(
     the remaining headers.
     """
     iter_headers = iter(headers)
+    # for logging:
+    first_discarded = None
+    last_discarded = None
+    num_discarded = 0
     for header in iter_headers:
         is_present = await completion_check(header)
         if is_present:
-            logger.debug("Discarding header that we already have: %s", header)
+            if first_discarded is None:
+                first_discarded = header
+            else:
+                last_discarded = header
+            num_discarded += 1
         else:
             yield header
             break
 
+    logger.debug(
+        "Discarding %d headers that we already have: %s...%s",
+        num_discarded,
+        first_discarded,
+        last_discarded,
+    )
+
     for header in iter_headers:
         yield header

trinity/sync/common/chain.py

@@ -41,6 +41,9 @@
     BaseService,
 )
 
+from trinity._utils.headers import (
+    skip_complete_headers,
+)
 from trinity.chains.base import BaseAsyncChain
 from trinity.db.eth1.header import BaseAsyncHeaderDB
 from trinity.protocol.common.peer import (
@@ -109,7 +112,7 @@ async def next_header_batch(self) -> AsyncIterator[Tuple[BlockHeader, ...]]:
         # When we start the sync with a peer, we always request up to MAX_REORG_DEPTH extra
         # headers before our current head's number, in case there were chain reorgs since the last
         # time _sync() was called. All of the extra headers that are already present in our DB
-        # will be discarded by _fetch_missing_headers() so we don't unnecessarily process them
+        # will be discarded by skip_complete_headers() so we don't unnecessarily process them
         # again.
         start_at = max(GENESIS_BLOCK_NUMBER + 1, head.block_number - MAX_REORG_DEPTH)
         while self.is_operational:
@@ -121,12 +124,10 @@ async def next_header_batch(self) -> AsyncIterator[Tuple[BlockHeader, ...]]:
                 all_headers = await self.wait(self._request_headers(peer, start_at))
                 if last_received_header is None:
                     # Skip over existing headers on the first run-through
-                    headers = tuple(
-                        # The inner list comprehension is needed because async_generators
-                        # cannot be cast to a tuple.
-                        [header async for header in self._get_missing_tail(all_headers)]
+                    new_headers = await self.wait(
+                        skip_complete_headers(all_headers, self.logger, self.db.coro_header_exists)
                     )
-                    if len(headers) == 0 and len(all_headers) > 0:
+                    if len(new_headers) == 0 and len(all_headers) > 0:
                         head = await self.wait(self.db.coro_get_canonical_head())
                         start_at = max(
                             all_headers[-1].block_number + 1,
@@ -140,8 +141,8 @@ async def next_header_batch(self) -> AsyncIterator[Tuple[BlockHeader, ...]]:
                         )
                         continue
                 else:
-                    headers = all_headers
-                self.logger.debug2('sync received new headers: %s', headers)
+                    new_headers = all_headers
+                self.logger.debug2('sync received new headers: %s', new_headers)
             except OperationCancelled:
                 self.logger.info("Sync with %s completed", peer)
                 break
@@ -157,7 +158,7 @@ async def next_header_batch(self) -> AsyncIterator[Tuple[BlockHeader, ...]]:
                 await peer.disconnect(DisconnectReason.useless_peer)
                 break
 
-            if not headers:
+            if not new_headers:
                 if last_received_header is None:
                     request_parent = head
                 else:
@@ -177,7 +178,7 @@ async def next_header_batch(self) -> AsyncIterator[Tuple[BlockHeader, ...]]:
                     self.logger.info("Got no new headers from %s, aborting sync", peer)
                 break
 
-            first = headers[0]
+            first = new_headers[0]
             first_parent = None
             if last_received_header is None:
                 # on the first request, make sure that the earliest ancestor has a parent in our db
@@ -205,27 +206,27 @@ async def next_header_batch(self) -> AsyncIterator[Tuple[BlockHeader, ...]]:
                 "Got new header chain from %s: %s..%s",
                 peer,
                 first,
-                headers[-1],
+                new_headers[-1],
             )
             try:
                 await self.chain.coro_validate_chain(
                     last_received_header or first_parent,
-                    headers,
+                    new_headers,
                     self._seal_check_random_sample_rate,
                 )
             except ValidationError as e:
                 self.logger.warning("Received invalid headers from %s, disconnecting: %s", peer, e)
                 await peer.disconnect(DisconnectReason.subprotocol_error)
                 break
 
-            for header in headers:
+            for header in new_headers:
                 head_td += header.difficulty
 
             # Setting the latest header hash for the peer, before queuing header processing tasks
             self._target_header_hash = peer.head_hash
 
-            yield headers
-            last_received_header = headers[-1]
+            yield new_headers
+            last_received_header = new_headers[-1]
             self.sync_progress = self.sync_progress.update_current_block(
                 last_received_header.block_number,
             )
@@ -243,26 +244,6 @@ async def _request_headers(
             reverse=False,
         )
 
-    async def _get_missing_tail(
-            self,
-            headers: Tuple[BlockHeader, ...]) -> AsyncIterator[BlockHeader]:
-        """
-        We only want headers that are missing, so we iterate over the list
-        until we find the first missing header, after which we return all of
-        the remaining headers.
-        """
-        iter_headers = iter(headers)
-        for header in iter_headers:
-            is_present = await self.wait(self.db.coro_header_exists(header.hash))
-            if is_present:
-                self.logger.debug("Discarding header that we already have: %s", header)
-            else:
-                yield header
-                break
-
-        for header in iter_headers:
-            yield header
-
 
 class BaseBlockImporter(ABC):
     @abstractmethod