Establish repeatable pattern for request response handling

Author: pipermerriam

p2p/peer.py

@@ -155,14 +155,6 @@ class BasePeer(BaseService):
     head_td: int = None
     head_hash: Hash32 = None
 
-    # TODO: Instead of a fixed timeout, we should instead monitor response
-    # times for the peer and adjust our timeout accordingly
-    _response_timeout = 60
-    pending_requests: Dict[
-        Type[protocol.Command],
-        Tuple['BaseRequest', 'asyncio.Future[protocol._DecodedMsgType]'],
-    ]
-
     def __init__(self,
                  remote: Node,
                  privkey: datatypes.PrivateKey,
@@ -189,8 +181,6 @@ def __init__(self,
         self.start_time = datetime.datetime.now()
         self.received_msgs: Dict[protocol.Command, int] = collections.defaultdict(int)
 
-        self.pending_requests = {}
-
         self.egress_mac = egress_mac
         self.ingress_mac = ingress_mac
         # FIXME: Yes, the encryption is insecure, see: https://github.com/ethereum/devp2p/issues/32
@@ -400,22 +390,6 @@ def handle_sub_proto_msg(self, cmd: protocol.Command, msg: protocol._DecodedMsgT
         else:
             self.logger.warn("Peer %s has no subscribers, discarding %s msg", self, cmd)
 
-        if cmd_type in self.pending_requests:
-            request, future = self.pending_requests[cmd_type]
-            try:
-                request.validate_response(msg)
-            except ValidationError as err:
-                self.logger.debug(
-                    "Response validation failure for pending %s request from peer %s: %s",
-                    cmd_type.__name__,
-                    self,
-                    err,
-                )
-                pass
-            else:
-                future.set_result(msg)
-                self.pending_requests.pop(cmd_type)
-
     def process_msg(self, cmd: protocol.Command, msg: protocol._DecodedMsgType) -> None:
         if cmd.is_base_protocol:
             self.handle_p2p_msg(cmd, msg)
@@ -640,6 +614,14 @@ def subscribe(self, peer_pool: 'PeerPool') -> Iterator[None]:
         finally:
             peer_pool.unsubscribe(self)
 
+    @contextlib.contextmanager
+    def subscribe_peer(self, peer: BasePeer) -> Iterator[None]:
+        peer.add_subscriber(self)
+        try:
+            yield
+        finally:
+            peer.remove_subscriber(self)
+
 
 class PeerPool(BaseService, AsyncIterable[BasePeer]):
     """
@@ -794,7 +776,7 @@ async def ensure_same_side_on_dao_fork(
         wait for that we may receive other messages from the peer, which are returned so that they
         can be re-added to our subscribers' queues when the peer is finally added to the pool.
         """
-        from trinity.protocol.base_block_headers import BaseBlockHeaders
+        from trinity.protocol.common.commands import BaseBlockHeaders
         msgs = []
         for start_block, vm_class in self.vm_configuration:
             if not issubclass(vm_class, HomesteadVM):
@@ -1008,7 +990,9 @@ def _test() -> None:
     from eth.chains.ropsten import RopstenChain, ROPSTEN_GENESIS_HEADER, ROPSTEN_VM_CONFIGURATION
     from eth.db.backends.memory import MemoryDB
     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.p2p.integration_test_helpers import FakeAsyncHeaderDB, connect_to_peers_loop
     logging.basicConfig(level=TRACE_LEVEL_NUM, format='%(asctime)s %(levelname)s: %(message)s')
 
@@ -1041,13 +1025,15 @@ async def request_stuff() -> None:
             '0x59af08ab31822c992bb3dad92ddb68d820aa4c69e9560f07081fa53f1009b152')
         if peer_class == ETHPeer:
             peer = cast(ETHPeer, peer)
-            peer.sub_proto.send_get_block_headers(block_hash, 1, 0, False)
+            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])
         else:
             peer = cast(LESPeer, peer)
             request_id = 1
-            peer.sub_proto.send_get_block_headers(block_hash, 1, 0, False, request_id)
+            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)
 

p2p/service.py

@@ -94,7 +94,6 @@ async def cleanup(self) -> None:
         The ``_cleanup()`` coroutine is invoked before the child services may have finished
         their cleanup.
         """
-
         await asyncio.gather(*[
             child_service.cleaned_up.wait()
             for child_service in self._child_services],

tests/conftest.py

@@ -33,7 +33,7 @@ def vm_logger(namespaces=LOGGING_NAMESPACES):
 
         handler = logging.StreamHandler(sys.stdout)
 
-        # level = TRACE_LEVEL_NUM
+        # level = 5  # TRACE
         # level = logging.DEBUG
         # level = logging.INFO
         level = logging.ERROR

tests/p2p/test_peer_subscriber.py

@@ -8,6 +8,7 @@
 
 from trinity.protocol.eth.peer import ETHPeer
 from trinity.protocol.eth.commands import GetBlockHeaders
+from trinity.protocol.eth.requests import HeaderRequest
 
 from tests.trinity.core.peer_helpers import (
     get_directly_linked_peers,
@@ -45,9 +46,9 @@ async def test_peer_subscriber_filters_messages(request, event_loop):
     peer.add_subscriber(all_subscriber)
 
     remote.sub_proto.send_get_node_data([b'\x00' * 32])
-    remote.sub_proto.send_get_block_headers(0, 1, 0, False)
+    remote.sub_proto.send_get_block_headers(HeaderRequest(0, 1, 0, False))
     remote.sub_proto.send_get_node_data([b'\x00' * 32])
-    remote.sub_proto.send_get_block_headers(1, 1, 0, False)
+    remote.sub_proto.send_get_block_headers(HeaderRequest(1, 1, 0, False))
     remote.sub_proto.send_get_node_data([b'\x00' * 32])
 
     # yeild to let remote and peer transmit.

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

@@ -2,7 +2,7 @@
 
 from p2p.exceptions import ValidationError
 
-from trinity.protocol.base_request import BaseHeaderRequest
+from trinity.protocol.common.requests import BaseHeaderRequest
 
 
 FORWARD_0_to_5 = (0, 6, 0, False)

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

@@ -72,10 +72,9 @@ async def test_eth_peer_get_headers_round_trip(eth_peer_and_remote,
 
     async def send_headers():
         remote.sub_proto.send_block_headers(headers)
-        await asyncio.sleep(0)
 
     asyncio.ensure_future(send_headers())
-    response = await peer.get_block_headers(*params)
+    response = await peer.handler.get_block_headers(*params)
 
     assert len(response) == len(headers)
     for expected, actual in zip(headers, response):
@@ -95,16 +94,14 @@ async def test_les_peer_get_headers_round_trip(les_peer_and_remote,
                                                params,
                                                headers):
     peer, remote = les_peer_and_remote
-    request_id = 1234
-
-    peer.gen_request_id = lambda: request_id
 
     async def send_headers():
+        request_id = peer.handler.get_block_headers.pending_request[0].request_id
         remote.sub_proto.send_block_headers(headers, 0, request_id)
         await asyncio.sleep(0)
 
     asyncio.ensure_future(send_headers())
-    response = await peer.get_block_headers(*params)
+    response = await peer.handler.get_block_headers(*params)
 
     assert len(response) == len(headers)
     for expected, actual in zip(headers, response):
@@ -124,7 +121,7 @@ async def send_responses():
         await asyncio.sleep(0)
 
     asyncio.ensure_future(send_responses())
-    response = await peer.get_block_headers(0, 10, 0, False)
+    response = await peer.handler.get_block_headers(0, 10, 0, False)
 
     assert len(response) == len(headers)
     for expected, actual in zip(headers, response):
@@ -148,7 +145,7 @@ async def send_responses():
         await asyncio.sleep(0)
 
     asyncio.ensure_future(send_responses())
-    response = await peer.get_block_headers(0, 5, 0, False)
+    response = await peer.handler.get_block_headers(0, 5, 0, False)
 
     assert len(response) == len(headers)
     for expected, actual in zip(headers, response):

trinity/exceptions.py

@@ -22,3 +22,11 @@ class MissingPath(BaseTrinityError):
     def __init__(self, msg: str, path: pathlib.Path) -> None:
         super().__init__(msg)
         self.path = path
+
+
+class AlreadyWaiting(BaseTrinityError):
+    """
+    Raised when an attempt is made to wait for a certain message type from a
+    peer when there is already an active wait for that message type.
+    """
+    pass

trinity/p2p/handlers.py

@@ -22,7 +22,7 @@
 from trinity.db.chain import AsyncChainDB
 from trinity.db.header import AsyncHeaderDB
 from trinity.protocol.eth.peer import ETHPeer
-from trinity.protocol.base_request import BaseHeaderRequest
+from trinity.protocol.common.requests import BaseHeaderRequest
 from trinity.rlp.block_body import BlockBody