Make EventBus requests less wasteful

Author: cburgdorf

tests/trinity/conftest.py

@@ -13,6 +13,9 @@
     Chain,
 )
 
+from trinity.constants import (
+    NETWORKING_EVENTBUS_ENDPOINT,
+)
 from trinity.chains.coro import (
     AsyncChainMixin,
 )
@@ -67,7 +70,7 @@ def event_loop():
 @pytest.fixture(scope='module')
 async def event_bus(event_loop):
     bus = EventBus()
-    endpoint = bus.create_endpoint('test')
+    endpoint = bus.create_endpoint(NETWORKING_EVENTBUS_ENDPOINT)
     bus.start(event_loop)
     await endpoint.connect(event_loop)
     try:

trinity/constants.py

@@ -1,6 +1,10 @@
 from pathlib import Path
 from typing import Dict, Tuple
 
+from lahja import (
+    BroadcastConfig,
+)
+
 from eth_utils import (
     decode_hex,
 )
@@ -23,6 +27,7 @@
 # lahja endpoint names
 MAIN_EVENTBUS_ENDPOINT = 'main'
 NETWORKING_EVENTBUS_ENDPOINT = 'networking'
+TO_NETWORKING_BROADCAST_CONFIG = BroadcastConfig(filter_endpoint=NETWORKING_EVENTBUS_ENDPOINT)
 
 # Network IDs: https://ethereum.stackexchange.com/questions/17051/how-to-select-a-network-id-or-is-there-a-list-of-network-ids/17101#17101  # noqa: E501
 MAINNET_NETWORK_ID = 1

trinity/plugins/builtin/ethstats/ethstats_service.py

@@ -19,6 +19,7 @@
 )
 from trinity.constants import (
     SYNC_LIGHT,
+    TO_NETWORKING_BROADCAST_CONFIG,
 )
 from trinity.db.manager import (
     create_db_manager,
@@ -138,7 +139,10 @@ async def get_node_stats(self) -> EthstatsData:
         '''Getter for data that should be sent periodically.'''
         try:
             peer_count = (await self.wait(
-                self.context.event_bus.request(PeerCountRequest()),
+                self.context.event_bus.request(
+                    PeerCountRequest(),
+                    TO_NETWORKING_BROADCAST_CONFIG,
+                ),
                 timeout=0.5
             )).peer_count
         except TimeoutError:

trinity/plugins/builtin/light_peer_chain_bridge/light_peer_chain_bridge.py

@@ -33,6 +33,9 @@
     BaseService,
 )
 
+from trinity.constants import (
+    TO_NETWORKING_BROADCAST_CONFIG,
+)
 from trinity.utils.async_errors import (
     await_and_wrap_errors,
 )
@@ -230,23 +233,33 @@ def __init__(self, event_bus: Endpoint) -> None:
 
     async def coro_get_block_header_by_hash(self, block_hash: Hash32) -> BlockHeader:
         event = GetBlockHeaderByHashRequest(block_hash)
-        return self._pass_or_raise(await self.event_bus.request(event)).block_header
+        return self._pass_or_raise(
+            await self.event_bus.request(event, TO_NETWORKING_BROADCAST_CONFIG)
+        ).block_header
 
     async def coro_get_block_body_by_hash(self, block_hash: Hash32) -> BlockBody:
         event = GetBlockBodyByHashRequest(block_hash)
-        return self._pass_or_raise(await self.event_bus.request(event)).block_body
+        return self._pass_or_raise(
+            await self.event_bus.request(event, TO_NETWORKING_BROADCAST_CONFIG)
+        ).block_body
 
     async def coro_get_receipts(self, block_hash: Hash32) -> List[Receipt]:
         event = GetReceiptsRequest(block_hash)
-        return self._pass_or_raise(await self.event_bus.request(event)).receipts
+        return self._pass_or_raise(
+            await self.event_bus.request(event, TO_NETWORKING_BROADCAST_CONFIG)
+        ).receipts
 
     async def coro_get_account(self, block_hash: Hash32, address: Address) -> Account:
         event = GetAccountRequest(block_hash, address)
-        return self._pass_or_raise(await self.event_bus.request(event)).account
+        return self._pass_or_raise(
+            await self.event_bus.request(event, TO_NETWORKING_BROADCAST_CONFIG)
+        ).account
 
     async def coro_get_contract_code(self, block_hash: Hash32, address: Address) -> bytes:
         event = GetContractCodeRequest(block_hash, address)
-        return self._pass_or_raise(await self.event_bus.request(event)).bytez
+        return self._pass_or_raise(
+            await self.event_bus.request(event, TO_NETWORKING_BROADCAST_CONFIG)
+        ).bytez
 
     TResponse = TypeVar("TResponse", bound=BaseLightPeerChainResponse)
 

trinity/rpc/modules/eth.py

@@ -40,6 +40,9 @@
     BaseAccountDB
 )
 
+from trinity.constants import (
+    TO_NETWORKING_BROADCAST_CONFIG,
+)
 from trinity.chains.base import BaseAsyncChain
 from trinity.rpc.format import (
     block_to_dict,
@@ -274,7 +277,7 @@ class SyncProgress(TypedDict):
         highestBlock: BlockNumber
 
     async def syncing(self) -> Union[bool, SyncProgress]:
-        res = await self._event_bus.request(SyncingRequest())
+        res = await self._event_bus.request(SyncingRequest(), TO_NETWORKING_BROADCAST_CONFIG)
         if res.is_syncing:
             return {
                 "startingBlock": res.progress.starting_block,

trinity/rpc/modules/net.py

@@ -1,4 +1,5 @@
 from p2p.events import PeerCountRequest
+from trinity.constants import TO_NETWORKING_BROADCAST_CONFIG
 from trinity.nodes.events import NetworkIdRequest
 from trinity.rpc.modules import RPCModule
 
@@ -8,14 +9,20 @@ async def version(self) -> str:
         """
         Returns the current network ID.
         """
-        response = await self._event_bus.request(NetworkIdRequest())
+        response = await self._event_bus.request(
+            NetworkIdRequest(),
+            TO_NETWORKING_BROADCAST_CONFIG
+        )
         return str(response.network_id)
 
     async def peerCount(self) -> str:
         """
         Return the number of peers that are currently connected to the node
         """
-        response = await self._event_bus.request(PeerCountRequest())
+        response = await self._event_bus.request(
+            PeerCountRequest(),
+            TO_NETWORKING_BROADCAST_CONFIG
+        )
         return hex(response.peer_count)
 
     async def listening(self) -> bool: