Merge pull request #1205 from carver/exchange-patch

ExchangeHandler stats fix

Author: carver

p2p/protocol.py

@@ -1,4 +1,4 @@
-from abc import ABC, abstractmethod
+from abc import ABC
 import logging
 import struct
 from typing import (
@@ -135,17 +135,14 @@ class BaseRequest(ABC, Generic[TRequestPayload]):
     Must define command_payload during init. This is the data that will
     be sent to the peer with the request command.
     """
+    # Defined at init time, with specific parameters:
     command_payload: TRequestPayload
 
-    @property
-    @abstractmethod
-    def cmd_type(self) -> Type[Command]:
-        raise NotImplementedError
-
-    @property
-    @abstractmethod
-    def response_type(self) -> Type[Command]:
-        raise NotImplementedError
+    # Defined as class attributes in subclasses
+    # outbound command type
+    cmd_type: Type[Command]
+    # response command type
+    response_type: Type[Command]
 
 
 class Protocol:

tests/trinity/core/p2p-proto/test_stats.py

@@ -0,0 +1,117 @@
+import asyncio
+
+from eth_utils import to_tuple
+from eth.rlp.headers import BlockHeader
+from p2p.peer import (
+    PeerSubscriber,
+)
+import pytest
+
+from trinity.protocol.les.commands import GetBlockHeaders
+from trinity.protocol.eth.peer import ETHPeer
+from trinity.protocol.les.peer import LESPeer
+
+from tests.trinity.core.peer_helpers import (
+    get_directly_linked_peers,
+)
+
+
+class RequestIDMonitor(PeerSubscriber):
+    subscription_msg_types = {GetBlockHeaders}
+    msg_queue_maxsize = 100
+
+    async def next_request_id(self):
+        msg = await self.msg_queue.get()
+        return msg.payload['request_id']
+
+
+@to_tuple
+def mk_header_chain(length):
+    assert length >= 1
+    genesis = BlockHeader(difficulty=100, block_number=0, gas_limit=3000000)
+    yield genesis
+    parent = genesis
+    if length == 1:
+        return
+
+    for i in range(length - 1):
+        header = BlockHeader(
+            difficulty=100,
+            block_number=parent.block_number + 1,
+            parent_hash=parent.hash,
+            gas_limit=3000000,
+        )
+        yield header
+        parent = header
+
+
+@pytest.fixture
+async def eth_peer_and_remote(request, event_loop):
+    peer, remote = await get_directly_linked_peers(
+        request,
+        event_loop,
+        peer1_class=ETHPeer,
+        peer2_class=ETHPeer,
+    )
+    return peer, remote
+
+
+@pytest.fixture
+async def les_peer_and_remote(request, event_loop):
+    peer, remote = await get_directly_linked_peers(
+        request,
+        event_loop,
+        peer1_class=LESPeer,
+        peer2_class=LESPeer,
+    )
+    return peer, remote
+
+
+@pytest.mark.asyncio
+async def test_eth_get_headers_empty_stats(eth_peer_and_remote):
+    peer, remote = eth_peer_and_remote
+    stats = peer.requests.get_stats()
+    assert all(status == 'Uninitialized' for status in stats.values())
+    assert 'BlockHeaders' in stats.keys()
+
+
+@pytest.mark.asyncio
+async def test_eth_get_headers_stats(eth_peer_and_remote):
+    peer, remote = eth_peer_and_remote
+
+    async def send_headers():
+        remote.sub_proto.send_block_headers(mk_header_chain(1))
+
+    for idx in range(1, 5):
+        get_headers_task = asyncio.ensure_future(peer.requests.get_block_headers(0, 1, 0, False))
+        asyncio.ensure_future(send_headers())
+
+        await get_headers_task
+
+        stats = peer.requests.get_stats()
+
+        assert stats['BlockHeaders'].startswith('count={0}, items={0}, avg_rtt='.format(idx))
+        assert stats['BlockHeaders'].endswith(', timeouts=0')
+
+
+@pytest.mark.asyncio
+async def test_les_get_headers_stats(les_peer_and_remote):
+    peer, remote = les_peer_and_remote
+
+    request_id_monitor = RequestIDMonitor()
+
+    for idx in range(1, 5):
+        with request_id_monitor.subscribe_peer(remote):
+            get_headers_task = asyncio.ensure_future(
+                peer.requests.get_block_headers(0, 1, 0, False)
+            )
+            request_id = await request_id_monitor.next_request_id()
+
+        remote.sub_proto.send_block_headers(mk_header_chain(1), 0, request_id)
+
+        await get_headers_task
+
+        stats = peer.requests.get_stats()
+
+        assert stats['BlockHeaders'].startswith('count={0}, items={0}, avg_rtt='.format(idx))
+        assert stats['BlockHeaders'].endswith(', timeouts=0')

trinity/protocol/common/exchanges.py

@@ -4,13 +4,16 @@
     Any,
     Callable,
     Generic,
+    Type,
 )
 
 from p2p.protocol import (
     BaseRequest,
+    Command,
     TRequestPayload,
 )
 
+from trinity.utils.decorators import classproperty
 from .managers import ExchangeManager
 from .normalizers import BaseNormalizer
 from .types import (
@@ -38,6 +41,8 @@ class BaseExchange(ABC, Generic[TRequestPayload, TResponsePayload, TResult]):
     TResult is the response data after normalization
     """
 
+    request_class: Type[BaseRequest[TRequestPayload]]
+
     def __init__(self, mgr: ExchangeManager[TRequestPayload, TResponsePayload, TResult]) -> None:
         self._manager = mgr
 
@@ -56,7 +61,7 @@ async def get_result(
         - the payload validator is primed with the request payload
         """
         if not self._manager.is_running:
-            await self._manager.launch_service(request.response_type)
+            await self._manager.launch_service()
 
         # bind the outbound request payload to the payload validator
         message_validator = partial(payload_validator, request.command_payload)
@@ -69,9 +74,13 @@ async def get_result(
             timeout=timeout
         )
 
+    @classproperty
+    def response_cmd_type(cls) -> Type[Command]:
+        return cls.request_class.response_type
+
     @abstractmethod
     async def __call__(self, *args: Any, **kwargs: Any) -> None:
         """
         Issue the request to the peer for the desired data
         """
-        raise NotImplementedError()
+        raise NotImplementedError('__call__ must be defined on every Exchange')

trinity/protocol/common/handlers.py

@@ -2,7 +2,7 @@
 from typing import (
     Any,
     Dict,
-    List,
+    Set,
     Type,
 )
 
@@ -17,24 +17,28 @@
 
 
 class BaseExchangeHandler:
+    _exchange_managers: Set[ExchangeManager[Any, Any, Any]]
+
     @property
     @abstractmethod
     def _exchanges(self) -> Dict[str, Type[BaseExchange[Any, Any, Any]]]:
         pass
 
     def __init__(self, peer: BasePeer) -> None:
         self._peer = peer
+        self._exchange_managers = set()
 
         for attr, exchange_cls in self._exchanges.items():
             if hasattr(self, attr):
                 raise AttributeError(
                     "Unable to set manager on attribute `{0}` which is already "
                     "present on the class: {1}".format(attr, getattr(self, attr))
                 )
-            manager: ExchangeManager[Any, Any, Any] = ExchangeManager(self._peer, peer.cancel_token)
+            manager: ExchangeManager[Any, Any, Any]
+            manager = ExchangeManager(self._peer, exchange_cls.response_cmd_type, peer.cancel_token)
+            self._exchange_managers.add(manager)
             exchange = exchange_cls(manager)
             setattr(self, attr, exchange)
 
-    def get_stats(self) -> List[str]:
-        manager_attrs = self._exchanges.keys()
-        return [getattr(self, attr).get_stats() for attr in manager_attrs]
+    def get_stats(self) -> Dict[str, str]:
+        return dict(exchange_manager.get_stats() for exchange_manager in self._exchange_managers)

trinity/protocol/common/managers.py

@@ -69,7 +69,7 @@ class ResponseCandidateStream(
     #
     @property
     def subscription_msg_types(self) -> Set[Type[Command]]:
-        return {self._response_msg_type}
+        return {self.response_msg_type}
 
     msg_queue_maxsize = 100
 
@@ -87,7 +87,7 @@ def __init__(
         super().__init__(token)
         self._peer = peer
         self.response_times = ResponseTimeTracker()
-        self._response_msg_type = response_msg_type
+        self.response_msg_type = response_msg_type
 
     async def payload_candidates(
             self,
@@ -105,7 +105,7 @@ async def payload_candidates(
 
     @property
     def response_msg_name(self) -> str:
-        return self._response_msg_type.__name__
+        return self.response_msg_type.__name__
 
     def complete_request(self, item_count: int) -> None:
         self.pending_request = None
@@ -123,7 +123,7 @@ async def _run(self) -> None:
                 if peer != self._peer:
                     self.logger.error("Unexpected peer: %s  expected: %s", peer, self._peer)
                     continue
-                elif isinstance(cmd, self._response_msg_type):
+                elif isinstance(cmd, self.response_msg_type):
                     await self._handle_msg(cast(TResponsePayload, msg))
                 else:
                     self.logger.warning("Unexpected payload type: %s", cmd.__class__.__name__)
@@ -176,8 +176,8 @@ def _request(self, request: BaseRequest[TRequestPayload]) -> None:
     def _is_pending(self) -> bool:
         return self.pending_request is not None
 
-    def get_stats(self) -> str:
-        return '%s: %s' % (self.response_msg_name, self.response_times.get_stats())
+    def get_stats(self) -> Tuple[str, str]:
+        return (self.response_msg_name, self.response_times.get_stats())
 
 
 class ExchangeManager(Generic[TRequestPayload, TResponsePayload, TResult]):
@@ -186,14 +186,16 @@ class ExchangeManager(Generic[TRequestPayload, TResponsePayload, TResult]):
     def __init__(
             self,
             peer: BasePeer,
+            listening_for: Type[Command],
             cancel_token: CancelToken) -> None:
         self._peer = peer
         self._cancel_token = cancel_token
+        self._response_command_type = listening_for
 
-    async def launch_service(self, listening_for: Type[Command]) -> None:
+    async def launch_service(self) -> None:
         self._response_stream = ResponseCandidateStream(
             self._peer,
-            listening_for,
+            self._response_command_type,
             self._cancel_token,
         )
         self._peer.run_daemon(self._response_stream)
@@ -248,5 +250,8 @@ def service(self) -> BaseService:
         """
         return self._response_stream
 
-    def get_stats(self) -> str:
-        return self._response_stream.get_stats()
+    def get_stats(self) -> Tuple[str, str]:
+        if self._response_stream is None:
+            return (self._response_command_type.__name__, 'Uninitialized')
+        else:
+            return self._response_stream.get_stats()

trinity/protocol/eth/exchanges.py

@@ -54,6 +54,7 @@
 
 class GetBlockHeadersExchange(BaseGetBlockHeadersExchange):
     _normalizer = NoopNormalizer[Tuple[BlockHeader, ...]]()
+    request_class = GetBlockHeadersRequest
 
     async def __call__(  # type: ignore
             self,
@@ -65,7 +66,7 @@ async def __call__(  # type: ignore
 
         original_request_args = (block_number_or_hash, max_headers, skip, reverse)
         validator = GetBlockHeadersValidator(*original_request_args)
-        request = GetBlockHeadersRequest(*original_request_args)
+        request = self.request_class(*original_request_args)
 
         return await self.get_result(
             request,
@@ -81,21 +82,23 @@ async def __call__(  # type: ignore
 
 class GetNodeDataExchange(BaseNodeDataExchange):
     _normalizer = GetNodeDataNormalizer()
+    request_class = GetNodeDataRequest
 
     async def __call__(self, node_hashes: Tuple[Hash32, ...]) -> NodeDataBundles:  # type: ignore
         validator = GetNodeDataValidator(node_hashes)
-        request = GetNodeDataRequest(node_hashes)
+        request = self.request_class(node_hashes)
         return await self.get_result(request, self._normalizer, validator, noop_payload_validator)
 
 
 class GetReceiptsExchange(BaseExchange[Tuple[Hash32, ...], ReceiptsByBlock, ReceiptsBundles]):
     _normalizer = ReceiptsNormalizer()
+    request_class = GetReceiptsRequest
 
     async def __call__(self, headers: Tuple[BlockHeader, ...]) -> ReceiptsBundles:  # type: ignore
         validator = ReceiptsValidator(headers)
 
         block_hashes = tuple(header.hash for header in headers)
-        request = GetReceiptsRequest(block_hashes)
+        request = self.request_class(block_hashes)
 
         return await self.get_result(request, self._normalizer, validator, noop_payload_validator)
 
@@ -109,11 +112,12 @@ async def __call__(self, headers: Tuple[BlockHeader, ...]) -> ReceiptsBundles:
 
 class GetBlockBodiesExchange(BaseGetBlockBodiesExchange):
     _normalizer = GetBlockBodiesNormalizer()
+    request_class = GetBlockBodiesRequest
 
     async def __call__(self, headers: Tuple[BlockHeader, ...]) -> BlockBodyBundles:  # type: ignore
         validator = GetBlockBodiesValidator(headers)
 
         block_hashes = tuple(header.hash for header in headers)
-        request = GetBlockBodiesRequest(block_hashes)
+        request = self.request_class(block_hashes)
 
         return await self.get_result(request, self._normalizer, validator, noop_payload_validator)

trinity/protocol/eth/peer.py

@@ -31,7 +31,8 @@ class ETHPeer(BasePeer):
     _requests: ETHExchangeHandler = None
 
     def get_extra_stats(self) -> List[str]:
-        return self.requests.get_stats()
+        stats_pairs = self.requests.get_stats().items()
+        return ['%s: %s' % (cmd_name, stats) for cmd_name, stats in stats_pairs]
 
     @property
     def requests(self) -> ETHExchangeHandler: