Track and report peer response size/time (#1166)

Closes: #1128

Author: gsalgado

p2p/peer.py

@@ -201,6 +201,9 @@ def __init__(self,
         mac_cipher = Cipher(algorithms.AES(mac_secret), modes.ECB(), default_backend())
         self.mac_enc = mac_cipher.encryptor().update
 
+    def get_extra_stats(self) -> List[str]:
+        return []
+
     @abstractmethod
     async def send_sub_proto_handshake(self) -> None:
         raise NotImplementedError("Must be implemented by subclasses")
@@ -916,12 +919,18 @@ async def _periodically_report_stats(self) -> None:
 
             self.logger.debug("== Peer details == ")
             for peer in self.connected_nodes.values():
+                if not peer.is_running:
+                    self.logger.warning(
+                        "%s is no longer alive but has not been removed from pool", peer)
+                    continue
                 most_received_type, count = max(
                     peer.received_msgs.items(), key=operator.itemgetter(1))
                 self.logger.debug(
-                    "%s: running=%s, uptime=%s, received_msgs=%d, most_received=%s(%d)",
-                    peer, peer.is_running, peer.uptime, peer.received_msgs_count,
+                    "%s: uptime=%s, received_msgs=%d, most_received=%s(%d)",
+                    peer, peer.uptime, peer.received_msgs_count,
                     most_received_type, count)
+                for line in peer.get_extra_stats():
+                    self.logger.debug("    %s", line)
             self.logger.debug("== End peer details == ")
             try:
                 await self.wait(asyncio.sleep(self._report_interval))
@@ -1018,9 +1027,7 @@ def _test() -> None:
     from eth.db.backends.memory import MemoryDB
     from eth.tools.logging import TRACE_LEVEL_NUM
     from trinity.protocol.eth.peer import ETHPeer
-    from trinity.protocol.eth.requests import HeaderRequest as ETHHeaderRequest
     from trinity.protocol.les.peer import LESPeer
-    from trinity.protocol.les.requests import HeaderRequest as LESHeaderRequest
     from tests.trinity.core.integration_test_helpers import FakeAsyncHeaderDB, connect_to_peers_loop
     logging.basicConfig(level=TRACE_LEVEL_NUM, format='%(asctime)s %(levelname)s: %(message)s')
 
@@ -1056,21 +1063,17 @@ async def request_stuff() -> None:
             peer_pool.logger.info("Waiting for peer connection...")
             await asyncio.sleep(0.2)
         peer = peer_pool.highest_td_peer
-        block_hash = decode_hex(
-            '0x59af08ab31822c992bb3dad92ddb68d820aa4c69e9560f07081fa53f1009b152')
+        headers = await peer.requests.get_block_headers(2440319, max_headers=100)  # type: ignore
+        hashes = [header.hash for header in headers]
         if peer_class == ETHPeer:
             peer = cast(ETHPeer, peer)
-            peer.sub_proto.send_get_block_headers(ETHHeaderRequest(block_hash, 1, 0, False))
-            peer.sub_proto.send_get_block_bodies([block_hash])
-            peer.sub_proto.send_get_receipts([block_hash])
+            peer.sub_proto.send_get_block_bodies(hashes)
+            peer.sub_proto.send_get_receipts(hashes)
         else:
             peer = cast(LESPeer, peer)
             request_id = 1
-            peer.sub_proto.send_get_block_headers(
-                LESHeaderRequest(block_hash, 1, 0, False, request_id)
-            )
-            peer.sub_proto.send_get_block_bodies([block_hash], request_id + 1)
-            peer.sub_proto.send_get_receipts(block_hash, request_id + 2)
+            peer.sub_proto.send_get_block_bodies(hashes, request_id + 1)
+            peer.sub_proto.send_get_receipts(hashes[0], request_id + 2)
 
     sigint_received = asyncio.Event()
     for sig in [signal.SIGINT, signal.SIGTERM]:

p2p/protocol.py

@@ -83,7 +83,7 @@ def decode_payload(self, rlp_data: bytes) -> _DecodedMsgType:
             decoder = sedes.List(
                 [type_ for _, type_ in self.structure], strict=self.decode_strict)
         try:
-            data = rlp.decode(rlp_data, sedes=decoder)
+            data = rlp.decode(rlp_data, sedes=decoder, recursive_cache=True)
         except rlp.DecodingError as err:
             raise MalformedMessage(
                 "Malformed %s message: %r".format(type(self).__name__, err)

trinity/protocol/common/handlers.py

@@ -2,6 +2,7 @@
 from typing import (
     Any,
     Dict,
+    List,
     Type,
 )
 
@@ -39,3 +40,7 @@ async def _run(self) -> None:
 
     async def _cleanup(self) -> None:
         pass
+
+    def get_stats(self) -> List[str]:
+        manager_attrs = self._managers.keys()
+        return [getattr(self, attr).get_stats() for attr in manager_attrs]

trinity/protocol/common/managers.py

@@ -1,5 +1,6 @@
 from abc import abstractmethod
 import asyncio
+import time
 from typing import (
     cast,
     Generic,
@@ -33,6 +34,31 @@
 TReturn = TypeVar('TReturn')
 
 
+class ResponseTimeTracker:
+
+    def __init__(self) -> None:
+        self.total_msgs = 0
+        self.total_items = 0
+        self.total_timeouts = 0
+        self.total_response_time = 0.0
+
+    def get_stats(self) -> str:
+        if not self.total_msgs:
+            return 'None'
+        avg_rtt = self.total_response_time / self.total_msgs
+        if not self.total_items:
+            per_item_rtt = 0.0
+        else:
+            per_item_rtt = self.total_response_time / self.total_items
+        return 'count=%d, items=%d, avg_rtt=%.2f, avg_time_per_item=%.5f, timeouts=%d' % (
+            self.total_msgs, self.total_items, avg_rtt, per_item_rtt, self.total_timeouts)
+
+    def add(self, time: float, size: int) -> None:
+        self.total_msgs += 1
+        self.total_items += size
+        self.total_response_time += time
+
+
 class BaseRequestManager(PeerSubscriber, BaseService, Generic[TPeer, TRequest, TResponse, TReturn]):  # noqa: E501
     #
     # PeerSubscriber
@@ -51,6 +77,7 @@ def subscription_msg_types(self) -> Set[Type[Command]]:
 
     def __init__(self, peer: TPeer, token: CancelToken) -> None:
         self._peer = peer
+        self.response_times = ResponseTimeTracker()
         super().__init__(token)
 
     #
@@ -61,8 +88,7 @@ async def _run(self) -> None:
 
         with self.subscribe_peer(self._peer):
             while self.is_running:
-                peer, cmd, msg = await self.wait(
-                    self.msg_queue.get(), token=self.cancel_token)
+                peer, cmd, msg = await self.wait(self.msg_queue.get())
                 if peer != self._peer:
                     self.logger.error("Unexpected peer: %s  expected: %s", peer, self._peer)
                     continue
@@ -81,12 +107,15 @@ async def _handle_msg(self, msg: TResponse) -> None:
             )
             return
 
+        self.response_times.add(
+            time.time() - self._pending_request_start, self._get_item_count(msg))
+
         request, future = self.pending_request
 
         try:
             response = await self._normalize_response(msg)
         except MalformedMessage as err:
-            self.logger.warn(
+            self.logger.warning(
                 "Malformed response for pending %s request from peer %s, disconnecting: %s",
                 self.response_msg_name,
                 self._peer,
@@ -112,6 +141,10 @@ async def _handle_msg(self, msg: TResponse) -> None:
     async def _normalize_response(self, msg: TResponse) -> TReturn:
         pass
 
+    @abstractmethod
+    def _get_item_count(self, msg: TResponse) -> int:
+        pass
+
     @abstractmethod
     def __call__(self) -> TReturn:
         """
@@ -141,7 +174,23 @@ def _send_sub_proto_request(self, request: TRequest) -> None:
 
     async def _wait_for_response(self,
                                  request: TRequest,
-                                 timeout: int = None) -> TReturn:
+                                 timeout: int) -> TReturn:
+        future: 'asyncio.Future[TReturn]' = asyncio.Future()
+        self._pending_request_start = time.time()
+        self.pending_request = (request, future)
+
+        try:
+            response = await self.wait(future, timeout=timeout)
+        except TimeoutError:
+            self.response_times.total_timeouts += 1
+            raise
+        finally:
+            # Always ensure that we reset the `pending_request` to `None` on exit.
+            self.pending_request = None
+
+        return response
+
+    async def _request_and_wait(self, request: TRequest, timeout: int=None) -> TReturn:
         if self.pending_request is not None:
             self.logger.error(
                 "Already waiting for response to %s for peer: %s",
@@ -155,19 +204,10 @@ async def _wait_for_response(self,
                 )
             )
 
-        future: 'asyncio.Future[TReturn]' = asyncio.Future()
-        self.pending_request = (request, future)
-
-        try:
-            response = await self.wait(future, timeout=timeout)
-        finally:
-            # Always ensure that we reset the `pending_request` to `None` on exit.
-            self.pending_request = None
-
-        return response
-
-    async def _request_and_wait(self, request: TRequest, timeout: int=None) -> TReturn:
         if timeout is None:
             timeout = self.response_timout
         self._send_sub_proto_request(request)
         return await self._wait_for_response(request, timeout=timeout)
+
+    def get_stats(self) -> str:
+        return '%s: %s' % (self.response_msg_name, self.response_times.get_stats())

trinity/protocol/eth/handlers.py

@@ -14,5 +14,6 @@ class ETHRequestResponseHandler(BaseRequestResponseHandler):
         'get_node_data': GetNodeDataRequestManager,
     }
 
+    # These are needed only to please mypy.
     get_block_headers: GetBlockHeadersRequestManager
     get_node_data: GetNodeDataRequestManager

trinity/protocol/eth/managers.py

@@ -73,6 +73,9 @@ async def _normalize_response(self,
                                   ) -> Tuple[BlockHeader, ...]:
         return msg
 
+    def _get_item_count(self, msg: Tuple[BlockHeader, ...]) -> int:
+        return len(msg)
+
 
 BaseGetNodeDataRequestManager = BaseRequestManager[
     'ETHPeer',
@@ -106,3 +109,6 @@ async def _normalize_response(self,
 
         node_keys = await self._run_in_executor(tuple, map(keccak, msg))
         return tuple(zip(node_keys, msg))
+
+    def _get_item_count(self, msg: Tuple[bytes, ...]) -> int:
+        return len(msg)

trinity/protocol/eth/peer.py

@@ -2,6 +2,7 @@
     Any,
     cast,
     Dict,
+    List,
 )
 
 from eth_utils import encode_hex
@@ -29,6 +30,9 @@ class ETHPeer(BasePeer):
 
     _requests: ETHRequestResponseHandler = None
 
+    def get_extra_stats(self) -> List[str]:
+        return self.requests.get_stats()
+
     @property
     def requests(self) -> ETHRequestResponseHandler:
         if self._requests is None:

trinity/protocol/les/managers.py

@@ -81,3 +81,6 @@ async def _normalize_response(self,
             )
 
         return msg['headers']
+
+    def _get_item_count(self, msg: Dict[str, Any]) -> int:
+        return len(msg['headers'])

trinity/protocol/les/peer.py

@@ -2,6 +2,7 @@
     Any,
     cast,
     Dict,
+    List,
 )
 
 from eth_utils import encode_hex
@@ -40,6 +41,9 @@ class LESPeer(BasePeer):
 
     _requests: LESRequestResponseHandler = None
 
+    def get_extra_stats(self) -> List[str]:
+        return self.requests.get_stats()
+
     @property
     def requests(self) -> LESRequestResponseHandler:
         if self._requests is None: