Improve logging output during chain sync (#1478)

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