Improve logging output during chain sync (#1478)

Author: pipermerriam

tests/trinity/core/humanize-utils/test_humanize_elapsed.py

@@ -0,0 +1,41 @@
+import pytest
+
+from trinity.utils.humanize import humanize_elapsed
+
+
+SECOND = 1
+MINUTE = 60
+HOUR = 60 * 60
+DAY = 24 * HOUR
+YEAR = 365 * DAY
+MONTH = YEAR // 12
+WEEK = 7 * DAY
+
+
+@pytest.mark.parametrize(
+    'seconds,expected',
+    (
+        (0, '0s'),
+        (1, '1s'),
+        (60, '1m'),
+        (61, '1m1s'),
+        (119, '1m59s'),
+        (HOUR, '1h'),
+        (HOUR + 1, '1h0m1s'),
+        (HOUR + MINUTE + 1, '1h1m1s'),
+        (DAY + HOUR, '1d1h'),
+        (DAY + HOUR + MINUTE, '1d1h1m'),
+        (DAY + MINUTE, '1d0h1m'),
+        (DAY + MINUTE + 1, '1d0h1m'),
+        (WEEK + DAY + HOUR, '1w1d1h'),
+        (WEEK + DAY + HOUR + MINUTE, '1w1d1h'),
+        (WEEK + DAY + HOUR + SECOND, '1w1d1h'),
+        (MONTH + WEEK + DAY, '1m1w1d'),
+        (MONTH + WEEK + DAY + HOUR, '1m1w1d'),
+        (YEAR + MONTH + WEEK, '1y1m1w'),
+        (YEAR + MONTH + WEEK + DAY, '1y1m1w'),
+    ),
+)
+def test_humanize_elapsed(seconds, expected):
+    actual = humanize_elapsed(seconds)
+    assert actual == expected

trinity/protocol/common/trackers.py

@@ -4,15 +4,14 @@
     Generic,
     Optional,
     TypeVar,
-    Union,
 )
 
-from eth_utils import ValidationError
 
 from p2p.protocol import (
     BaseRequest,
 )
 
+from trinity.utils.ema import EMA
 from trinity.utils.logging import HasTraceLogger
 from .constants import ROUND_TRIP_TIMEOUT
 from .types import (
@@ -23,39 +22,6 @@
 TRequest = TypeVar('TRequest', bound=BaseRequest[Any])
 
 
-class EMA:
-    """
-    Represents an exponential moving average.
-    https://en.wikipedia.org/wiki/Moving_average#Exponential_moving_average
-
-    Smoothing factor, or "alpha" of the exponential moving average.
-
-    - Closer to 0 gives you smoother, slower-to-update, data
-    - Closer to 1 gives you choppier, quicker-to-update, data
-
-    .. note::
-
-        A smoothing factor of 1 would completely ignore history whereas 0 would
-        completely ignore new data
-
-
-    The initial value is the starting value for the EMA
-    """
-    def __init__(self, initial_value: float, smoothing_factor: float) -> None:
-        self._value = initial_value
-        if 0 < smoothing_factor < 1:
-            self._alpha = smoothing_factor
-        else:
-            raise ValidationError("Smoothing factor of EMA must be between 0 and 1")
-
-    def update(self, scalar: Union[int, float]) -> None:
-        self._value = (self._value * (1 - self._alpha)) + (scalar * self._alpha)
-
-    @property
-    def value(self) -> float:
-        return self._value
-
-
 class BasePerformanceTracker(ABC, HasTraceLogger, Generic[TRequest, TResult]):
     def __init__(self) -> None:
         self.total_msgs = 0

trinity/sync/full/chain.py

@@ -3,6 +3,7 @@
     PriorityQueue,
 )
 from concurrent.futures import CancelledError
+import datetime
 import enum
 from functools import (
     partial,
@@ -11,6 +12,7 @@
 from typing import (
     Dict,
     List,
+    NamedTuple,
     Set,
     Tuple,
     Type,
@@ -62,6 +64,8 @@
     OrderedTaskPreparation,
     TaskQueue,
 )
+from trinity.utils.ema import EMA
+from trinity.utils.humanize import humanize_elapsed
 from trinity.utils.timer import Timer
 
 # (ReceiptBundle, (Receipt, (root_hash, receipt_trie_data))
@@ -338,6 +342,71 @@ class BlockPersistPrereqs(enum.Enum):
     StoreReceipts = enum.auto()
 
 
+class ChainSyncStats(NamedTuple):
+    prev_head: BlockHeader
+    latest_head: BlockHeader
+
+    elapsed: float
+
+    num_blocks: int
+    blocks_per_second: float
+
+    num_transactions: int
+    transactions_per_second: float
+
+
+class ChainSyncPerformanceTracker:
+    def __init__(self, head: BlockHeader) -> None:
+        # The `head` from the previous time we reported stats
+        self.prev_head = head
+        # The latest `head` we have synced
+        self.latest_head = head
+
+        # A `Timer` object to report elapsed time between reports
+        self.timer = Timer()
+
+        # EMA of the blocks per second
+        self.blocks_per_second_ema = EMA(initial_value=0, smoothing_factor=0.05)
+
+        # EMA of the transactions per second
+        self.transactions_per_second_ema = EMA(initial_value=0, smoothing_factor=0.05)
+
+        # Number of transactions processed
+        self.num_transactions = 0
+
+    def record_transactions(self, count: int) -> None:
+        self.num_transactions += count
+
+    def set_latest_head(self, head: BlockHeader) -> None:
+        self.latest_head = head
+
+    def report(self) -> ChainSyncStats:
+        elapsed = self.timer.pop_elapsed()
+
+        num_blocks = self.latest_head.block_number - self.prev_head.block_number
+        blocks_per_second = num_blocks / elapsed
+        transactions_per_second = self.num_transactions / elapsed
+
+        self.blocks_per_second_ema.update(blocks_per_second)
+        self.transactions_per_second_ema.update(transactions_per_second)
+
+        stats = ChainSyncStats(
+            prev_head=self.prev_head,
+            latest_head=self.latest_head,
+            elapsed=elapsed,
+            num_blocks=num_blocks,
+            blocks_per_second=self.blocks_per_second_ema.value,
+            num_transactions=self.num_transactions,
+            transactions_per_second=self.transactions_per_second_ema.value,
+        )
+
+        # reset the counters
+        self.num_transactions = 0
+        self.prev_head = self.latest_head
+
+        return stats
+
+
 class FastChainSyncer(BaseBodyChainSyncer):
     """
     Sync with the Ethereum network by fetching block headers/bodies and storing them in our DB.
@@ -374,6 +443,8 @@ def __init__(self,
 
     async def _run(self) -> None:
         head = await self.wait(self.db.coro_get_canonical_head())
+        self.tracker = ChainSyncPerformanceTracker(head)
+
         self._block_persist_tracker.set_finished_dependency(head)
         self.run_daemon_task(self._launch_prerequisite_tasks())
         self.run_daemon_task(self._assign_receipt_download_to_peers())
@@ -445,9 +516,6 @@ async def _launch_prerequisite_tasks(self) -> None:
             self.header_queue.complete(batch_id, headers)
 
     async def _display_stats(self) -> None:
-        last_head = await self.wait(self.db.coro_get_canonical_head())
-        timer = Timer()
-
         while self.is_operational:
             await self.sleep(5)
             self.logger.debug(
@@ -459,16 +527,29 @@ async def _display_stats(self) -> None:
                 )],
             )
 
-            head = await self.wait(self.db.coro_get_canonical_head())
-            if head == last_head:
-                continue
-            else:
-                block_num_change = head.block_number - last_head.block_number
-                last_head = head
-
-                self.logger.info(
-                    "Advanced by %d blocks in %0.1f seconds, new head: %s",
-                    block_num_change, timer.pop_elapsed(), head)
+            stats = self.tracker.report()
+            utcnow = int(datetime.datetime.utcnow().timestamp())
+            head_age = utcnow - stats.latest_head.timestamp
+            self.logger.info(
+                (
+                    "blks=%-4d  "
+                    "txs=%-5d  "
+                    "bps=%-3d  "
+                    "tps=%-4d  "
+                    "elapsed=%0.1f  "
+                    "head=#%d (%s\u2026%s)  "
+                    "age=%s"
+                ),
+                stats.num_blocks,
+                stats.num_transactions,
+                stats.blocks_per_second,
+                stats.transactions_per_second,
+                stats.elapsed,
+                stats.latest_head.block_number,
+                stats.latest_head.hex_hash[2:6],
+                stats.latest_head.hex_hash[-4:],
+                humanize_elapsed(head_age),
+            )
 
     async def _persist_ready_blocks(self) -> None:
         """
@@ -514,8 +595,12 @@ async def _persist_blocks(self, headers: Tuple[BlockHeader, ...]) -> None:
                 tx_class = block_class.get_transaction_class()
                 transactions = [tx_class.from_base_transaction(tx) for tx in body.transactions]
 
+                # record progress in the tracker
+                self.tracker.record_transactions(len(transactions))
+
             block = block_class(header, transactions, uncles)
             await self.wait(self.db.coro_persist_block(block))
+            self.tracker.set_latest_head(header)
 
     async def _assign_receipt_download_to_peers(self) -> None:
         """

trinity/utils/ema.py

@@ -0,0 +1,36 @@
+from typing import Union
+
+from eth_utils import ValidationError
+
+
+class EMA:
+    """
+    Represents an exponential moving average.
+    https://en.wikipedia.org/wiki/Moving_average#Exponential_moving_average
+
+    Smoothing factor, or "alpha" of the exponential moving average.
+
+    - Closer to 0 gives you smoother, slower-to-update, data
+    - Closer to 1 gives you choppier, quicker-to-update, data
+
+    .. note::
+
+        A smoothing factor of 1 would completely ignore history whereas 0 would
+        completely ignore new data
+
+
+    The initial value is the starting value for the EMA
+    """
+    def __init__(self, initial_value: float, smoothing_factor: float) -> None:
+        self._value = initial_value
+        if 0 < smoothing_factor < 1:
+            self._alpha = smoothing_factor
+        else:
+            raise ValidationError("Smoothing factor of EMA must be between 0 and 1")
+
+    def update(self, scalar: Union[int, float]) -> None:
+        self._value = (self._value * (1 - self._alpha)) + (scalar * self._alpha)
+
+    @property
+    def value(self) -> float:
+        return self._value

trinity/utils/humanize.py

@@ -0,0 +1,46 @@
+from typing import Iterator
+
+
+def humanize_elapsed(seconds: int) -> str:
+    return ''.join(_humanize_elapsed(seconds))
+
+
+SECOND = 1
+MINUTE = 60
+HOUR = 60 * 60
+DAY = 24 * HOUR
+YEAR = 365 * DAY
+MONTH = YEAR // 12
+WEEK = 7 * DAY
+
+
+UNITS = (
+    (YEAR, 'y'),
+    (MONTH, 'm'),
+    (WEEK, 'w'),
+    (DAY, 'd'),
+    (HOUR, 'h'),
+    (MINUTE, 'm'),
+    (SECOND, 's'),
+)
+
+
+def _humanize_elapsed(seconds: int) -> Iterator[str]:
+    if not seconds:
+        yield '0s'
+
+    num_display_units = 0
+    remainder = seconds
+
+    for duration, unit in UNITS:
+        if not remainder:
+            break
+        if remainder >= duration or num_display_units:
+            num = remainder // duration
+            yield f"{num}{unit}"
+            num_display_units += 1
+
+        if num_display_units >= 3:
+            return
+
+        remainder %= duration