Christoph/feat/plugins own process alternative

eth/chains/base.py

@@ -892,3 +892,15 @@ async def coro_validate_receipt(self,
                                     receipt: Receipt,
                                     at_header: BlockHeader) -> None:
         raise NotImplementedError()
+
+    async def coro_get_block_by_hash(self,
+                                     block_hash: Hash32) -> BaseBlock:
+        raise NotImplementedError()
+
+    async def coro_get_block_by_header(self,
+                                       header: BlockHeader) -> BaseBlock:
+        raise NotImplementedError()
+
+    async def coro_get_canonical_block_by_number(self,
+                                                 block_number: BlockNumber) -> BaseBlock:
+        raise NotImplementedError()

eth/tools/fixtures/helpers.py

@@ -146,12 +146,12 @@ def genesis_params_from_fixture(fixture):
     }
 
 
-def new_chain_from_fixture(fixture):
+def new_chain_from_fixture(fixture, chain_cls=MainnetChain):
     base_db = MemoryDB()
 
     vm_config = chain_vm_configuration(fixture)
 
-    ChainFromFixture = MainnetChain.configure(
+    ChainFromFixture = chain_cls.configure(
         'ChainFromFixture',
         vm_configuration=vm_config,
     )

p2p/events.py

@@ -0,0 +1,21 @@
+from typing import (
+    Type,
+)
+
+from lahja import (
+    BaseEvent,
+    BaseRequestResponseEvent,
+)
+
+
+class PeerCountResponse(BaseEvent):
+
+    def __init__(self, peer_count: int) -> None:
+        self.peer_count = peer_count
+
+
+class PeerCountRequest(BaseRequestResponseEvent[PeerCountResponse]):
+
+    @staticmethod
+    def expected_response_type() -> Type[PeerCountResponse]:
+        return PeerCountResponse

p2p/peer.py

@@ -52,6 +52,10 @@
 
 from cancel_token import CancelToken, OperationCancelled
 
+from lahja import (
+    Endpoint,
+)
+
 from eth.chains.mainnet import MAINNET_NETWORK_ID
 from eth.chains.ropsten import ROPSTEN_NETWORK_ID
 from eth.constants import GENESIS_BLOCK_NUMBER
@@ -101,6 +105,11 @@
     MAC_LEN,
 )
 
+from .events import (
+    PeerCountRequest,
+    PeerCountResponse,
+)
+
 if TYPE_CHECKING:
     from trinity.db.header import BaseAsyncHeaderDB  # noqa: F401
     from trinity.protocol.common.proto import ChainInfo  # noqa: F401
@@ -784,6 +793,7 @@ def __init__(self,
                  vm_configuration: Tuple[Tuple[int, Type[BaseVM]], ...],
                  max_peers: int = DEFAULT_MAX_PEERS,
                  token: CancelToken = None,
+                 event_bus: Endpoint = None
                  ) -> None:
         super().__init__(token)
         self.peer_class = peer_class
@@ -794,6 +804,16 @@ def __init__(self,
         self.max_peers = max_peers
         self.connected_nodes: Dict[Node, BasePeer] = {}
         self._subscribers: List[PeerSubscriber] = []
+        self.event_bus = event_bus
+        self.run_task(self.handle_peer_count_requests())
+
+    async def handle_peer_count_requests(self) -> None:
+        async for req in self.event_bus.stream(PeerCountRequest):
+            # We are listening for all `PeerCountRequest` events but we ensure to
+            # only send a `PeerCountResponse` to the callsite that made the request.
+            # We do that by retrieving a `BroadcastConfig` from the request via the
+            # `event.broadcast_config()` API.
+            self.event_bus.broadcast(PeerCountResponse(len(self)), req.broadcast_config())
 
     def __len__(self) -> int:
         return len(self.connected_nodes)

setup.py

@@ -42,7 +42,7 @@
         "ipython>=6.2.1,<7.0.0",
         "plyvel==1.0.5",
         "web3==4.4.1",
-        "lahja==0.6.1",
+        "lahja==0.8.0",
     ],
     'test': [
         "hypothesis==3.69.5",

tests/conftest.py

@@ -78,8 +78,7 @@ def funded_address_initial_balance():
     return to_wei(1000, 'ether')
 
 
-@pytest.fixture
-def chain_with_block_validation(base_db, genesis_state):
+def _chain_with_block_validation(base_db, genesis_state, chain_cls=Chain):
     """
     Return a Chain object containing just the genesis block.
 
@@ -107,7 +106,8 @@ def chain_with_block_validation(base_db, genesis_state):
         "transaction_root": decode_hex("56e81f171bcc55a6ff8345e692c0f86e5b48e01b996cadc001622fb5e363b421"),  # noqa: E501
         "uncles_hash": decode_hex("1dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d49347")  # noqa: E501
     }
-    klass = Chain.configure(
+
+    klass = chain_cls.configure(
         __name__='TestChain',
         vm_configuration=(
             (constants.GENESIS_BLOCK_NUMBER, SpuriousDragonVM),
@@ -118,6 +118,11 @@ def chain_with_block_validation(base_db, genesis_state):
     return chain
 
 
+@pytest.fixture
+def chain_with_block_validation(base_db, genesis_state):
+    return _chain_with_block_validation(base_db, genesis_state)
+
+
 def import_block_without_validation(chain, block):
     return super(type(chain), chain).import_block(block, perform_validation=False)
 

tests/trinity/conftest.py

@@ -5,8 +5,19 @@
 import tempfile
 import uuid
 
+from lahja import (
+    EventBus,
+)
+
+from eth.chains import (
+    Chain,
+)
+
 from p2p.peer import PeerPool
 
+from trinity.chains.coro import (
+    AsyncChainMixin,
+)
 from trinity.rpc.main import (
     RPCServer,
 )
@@ -22,6 +33,13 @@
 from trinity.utils.filesystem import (
     is_under_path,
 )
+from tests.conftest import (
+    _chain_with_block_validation,
+)
+
+
+class TestAsyncChain(Chain, AsyncChainMixin):
+    pass
 
 
 def pytest_addoption(parser):
@@ -51,6 +69,19 @@ def event_loop():
         loop.close()
 
 
+@pytest.fixture(scope='module')
+def event_bus(event_loop):
+    bus = EventBus()
+    endpoint = bus.create_endpoint('test')
+    bus.start(event_loop)
+    endpoint.connect(event_loop)
+    try:
+        yield endpoint
+    finally:
+        endpoint.stop()
+        bus.stop()
+
+
 @pytest.fixture(scope='session')
 def jsonrpc_ipc_pipe_path():
     with tempfile.TemporaryDirectory() as temp_dir:
@@ -64,11 +95,16 @@ def p2p_server(monkeypatch, jsonrpc_ipc_pipe_path):
     return Server(None, None, None, None, None, None, None)
 
 
+@pytest.fixture
+def chain_with_block_validation(base_db, genesis_state):
+    return _chain_with_block_validation(base_db, genesis_state, TestAsyncChain)
+
+
 @pytest.mark.asyncio
 @pytest.fixture
 async def ipc_server(
         monkeypatch,
-        p2p_server,
+        event_bus,
         jsonrpc_ipc_pipe_path,
         event_loop,
         chain_with_block_validation):
@@ -77,7 +113,7 @@ async def ipc_server(
     the course of all tests. It yields the IPC server only for monkeypatching purposes
     '''
 
-    rpc = RPCServer(chain_with_block_validation, p2p_server.peer_pool)
+    rpc = RPCServer(chain_with_block_validation, event_bus)
     ipc_server = IPCServer(rpc, jsonrpc_ipc_pipe_path, loop=event_loop)
 
     asyncio.ensure_future(ipc_server.run(), loop=event_loop)

tests/trinity/core/chain-management/test_light_peer_chain.py

@@ -0,0 +1,20 @@
+from trinity.sync.light.service import (
+    LightPeerChain
+)
+from trinity.plugins.builtin.light_peer_chain_bridge import (
+    EventBusLightPeerChain,
+)
+
+
+# These tests may seem obvious but they safe us from runtime errors where
+# changes are made to the `BaseLightPeerChain` that are then forgotton to
+# implement on both derived chains.
+
+def test_can_instantiate_eventbus_light_peer_chain():
+    chain = EventBusLightPeerChain(None)
+    assert chain is not None
+
+
+def test_can_instantiate_light_peer_chain():
+    chain = LightPeerChain(None, None)
+    assert chain is not None

tests/trinity/core/json-rpc/test_ipc.py

@@ -14,6 +14,11 @@
     to_hex,
 )
 
+from p2p.events import (
+    PeerCountRequest,
+    PeerCountResponse,
+)
+
 from trinity.utils.version import construct_trinity_client_identifier
 
 
@@ -35,15 +40,6 @@ def build_request(method, params=[]):
     return json.dumps(request).encode()
 
 
-class MockPeerPool:
-
-    def __init__(self, peer_count=0):
-        self.peer_count = peer_count
-
-    def __len__(self):
-        return self.peer_count
-
-
 def id_from_rpc_request(param):
     if isinstance(param, bytes):
         request = json.loads(param.decode())
@@ -68,6 +64,7 @@ async def get_ipc_response(
         jsonrpc_ipc_pipe_path,
         request_msg,
         event_loop):
+
     assert wait_for(jsonrpc_ipc_pipe_path), "IPC server did not successfully start with IPC file"
 
     reader, writer = await asyncio.open_unix_connection(str(jsonrpc_ipc_pipe_path), loop=event_loop)
@@ -399,18 +396,27 @@ async def test_eth_call_with_contract_on_ipc(
     assert result == expected
 
 
+def mock_peer_count(count):
+    async def mock_event_bus_interaction(bus):
+        async for req in bus.stream(PeerCountRequest):
+            bus.broadcast(PeerCountResponse(count), req.broadcast_config())
+            break
+
+    return mock_event_bus_interaction
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize(
-    'request_msg, mock_peer_pool, expected',
+    'request_msg, event_bus_setup_fn, expected',
     (
         (
             build_request('net_peerCount'),
-            MockPeerPool(peer_count=1),
+            mock_peer_count(1),
             {'result': '0x1', 'id': 3, 'jsonrpc': '2.0'},
         ),
         (
             build_request('net_peerCount'),
-            MockPeerPool(peer_count=0),
+            mock_peer_count(0),
             {'result': '0x0', 'id': 3, 'jsonrpc': '2.0'},
         ),
     ),
@@ -422,10 +428,17 @@ async def test_peer_pool_over_ipc(
         monkeypatch,
         jsonrpc_ipc_pipe_path,
         request_msg,
-        mock_peer_pool,
+        event_bus_setup_fn,
+        event_bus,
         expected,
         event_loop,
         ipc_server):
-    monkeypatch.setattr(ipc_server.rpc.modules['net'], '_peer_pool', mock_peer_pool)
-    result = await get_ipc_response(jsonrpc_ipc_pipe_path, request_msg, event_loop)
+
+    asyncio.ensure_future(event_bus_setup_fn(event_bus))
+
+    result = await get_ipc_response(
+        jsonrpc_ipc_pipe_path,
+        request_msg,
+        event_loop
+    )
     assert result == expected

tests/trinity/integration/test_lightchain_integration.py

@@ -198,26 +198,26 @@ async def wait_for_header_sync(block_number):
 
     # https://ropsten.etherscan.io/block/11
     header = headerdb.get_canonical_block_header_by_number(n)
-    body = await peer_chain.get_block_body_by_hash(header.hash)
+    body = await peer_chain.coro_get_block_body_by_hash(header.hash)
     assert len(body['transactions']) == 15
 
-    receipts = await peer_chain.get_receipts(header.hash)
+    receipts = await peer_chain.coro_get_receipts(header.hash)
     assert len(receipts) == 15
     assert encode_hex(keccak(rlp.encode(receipts[0]))) == (
         '0xf709ed2c57efc18a1675e8c740f3294c9e2cb36ba7bb3b89d3ab4c8fef9d8860')
 
     assert len(peer_pool) == 1
     peer = peer_pool.highest_td_peer
-    head = await peer_chain.get_block_header_by_hash(peer.head_hash)
+    head = await peer_chain.coro_get_block_header_by_hash(peer.head_hash)
 
     # In order to answer queries for contract code, geth needs the state trie entry for the block
     # we specify in the query, but because of fast sync we can only assume it has that for recent
     # blocks, so we use the current head to lookup the code for the contract below.
     # https://ropsten.etherscan.io/address/0x95a48dca999c89e4e284930d9b9af973a7481287
     contract_addr = decode_hex('0x8B09D9ac6A4F7778fCb22852e879C7F3B2bEeF81')
-    contract_code = await peer_chain.get_contract_code(head.hash, contract_addr)
+    contract_code = await peer_chain.coro_get_contract_code(head.hash, contract_addr)
     assert encode_hex(contract_code) == '0x600060006000600060006000356000f1'
 
-    account = await peer_chain.get_account(head.hash, contract_addr)
+    account = await peer_chain.coro_get_account(head.hash, contract_addr)
     assert account.code_hash == keccak(contract_code)
     assert account.balance == 0