Move RequestServer into isolated plugin #617

cburgdorf · 2019-05-16T13:38:18Z

What was wrong?

It is our long term plan to have the RequestServer, the PeerPool and the Sync run in their dedicated isolated processes. The benefits are:

The PeerPool will be free from any CPU intensive work (e.g. block verification etc)
Leverage multiple CPUs better
Have pieces be more reusable (e.g RequestServer plugin can be used for beacon node)
Have each process be a better performance analysis target via py-spy
???

Currently, these three things are still together in one process that is so far know as the networking process.

How was it fixed?

Upgraded to unreleased Lahja version that enables us to broadcast on a higher frequency (by reducing unnecessary cross process propagation when it isn't needed)
Exposed the relevant peer messages on the event bus
Created ProxyPeer classes that allow to forward actions to the actual peers in the peer pool
Created a new IsolatedRequestServer that consumes relevant peer pool messages via event bus and uses proxy peers to communicate back responses
Created a new isolated plugin that runs the isolated request server

To-Do

Clean up commit history

Add entry to the [release notes PR](https://github.com/ethereum/trinity/pulls?
q=is%3Aopen+is%3Apr+label%3A%22Release+Notes%22)
Run BCCRequestServer via the same plugin
Release new Lahja and update setup.py

Cute Animal Picture

pipermerriam

Approach looks workable. Lots of boilerplate in some areas but I think that's a product of our current deficiencies in our need for independent peer classes rather than any issue with your design.

pipermerriam · 2019-05-17T15:49:11Z

trinity/protocol/eth/peer.py

+                peer.sub_proto.send_block_headers(ev.headers)
+
+    async def handle_send_block_bodies_events(self) -> None:
+        async for ev in self.wait_iter(self.event_bus.stream(SendBlockBodiesEvent)):


These all look so similar. Maybe replace with

async def handle_event(self, StreamEventType, event_handler_fn): async for ev in self.wait_iter(self.event_bus.stream(StreamEventType)): try: peer = self.get_peer(ev.remote) except PeerConnectionLost: pass else: event_handler_fn(peer, ev)

Yep, I cleaned this up a bit more. Not sure what you think about the lambdas. I know you are not a fan but I couldn't figure out an alternative that I was satisfied with.

cburgdorf · 2019-05-20T13:49:36Z

@pipermerriam This can be given a full review now. It is still blocked on the upcoming lahja release but other than that it has passed my own judgement :)

carver

Just a quick scan, but LGTM. If you don't get a 👍 soon enough, ping me and I can take a second pass to approve.

carver · 2019-05-20T20:39:09Z

tests/core/integration_test_helpers.py

@@ -184,4 +188,18 @@ def load_fixture_db(db_fixture, db_class=LevelDB):
        peer_pool.cancel_token
    )
    asyncio.ensure_future(peer_pool_event_bus_request_handler.run())
+    # Give event subscriptions a moment to propagate
+    await asyncio.sleep(0.01)


I don't understand why this is necessary. Why isn't the next await enough?

carver · 2019-05-20T20:41:57Z

tests/core/json-rpc/test_ipc.py

@@ -87,6 +87,8 @@ def can_decode_json(potential):
        request_msg,
        event_loop):

+    # Give the event subscriptions a moment to propagate
+    await asyncio.sleep(0.01)


All these sleeps seem like a code smell. If they are required for correctness in non-test code, then we will probably have related bugs sooner or later.

We've got some wait_for_XX_subscripiton APIs coming down the pipe.

Check out the last commit. It gets rid of the sleeps but depends on ethereum/lahja#72 getting merged before.

Btw, I don't think that will bite us in real code. In general, the code should be written in a way that it can handle cases of the other end not being fully booted yet. E.g. we already have things like the PeerPool reaching out to the discovery to get peers but then timing out on the first request if the discovery isn't yet up and running to answer it.

I think this is more an issue of the test cases where we wire things together manually and have strict requirements about the other side being fully up and running yet.

I know this answer is kinda fuzzy and it might be wishful thinking but I think we are safe 😅

I was able to resolve all sleeps except this one with the existing wait_for APIs that are already in lahja/master.

For this one, we can not wait for a specific event but would rather like for any event subscription to be propagated. This would be possible with the upcoming wait_for_x APIs proposed in ethereum/lahja#74 but I would like to base this PR on top of lahja/master to land it sooner.

So I hope we can live with this one single sleep until we have figured out the best API to wait on subscription propagations.

Maybe add a comment with the proposed replacement code, to be used when the new API is merged/released?

The syntax isn't yet entirely finalized but I promise to come back to this and update it as soon as it lands!

Tracking issue?

tests/core/network-db/test_connection_tracker_server.py

carver · 2019-05-20T20:48:17Z

trinity/protocol/common/events.py

@@ -41,11 +54,24 @@ def expected_response_type() -> Type[PeerCountResponse]:
        return PeerCountResponse


-class DisconnectPeerEvent(BaseEvent):
+class DisconnectPeerEvent(HasRemoteEvent):
    """
    Event broadcasted when we want to disconnect from a peer
    """

    def __init__(self, remote: Node, reason: DisconnectReason) -> None:
        self.remote = remote


I guess this could also now be:

Suggested change

self.remote = remote

super().__init__(remote)

I'd advocate for dropping the HasRemoteEvent and just doing it explicitely. It doesn't appear to save any lines of code and I'd argue it makes readability worse.

def __init__(self, remote: Node, other_arg: int): super().__init__(remote) self.other_arg = other_arg # vs def __init__(self, remote: Node, other_arg: int): self.remote = remote self.other_arg = other_arg

We need that event for the bound in https://github.com/ethereum/trinity/pull/617/files#diff-8067591e62bc9b699e77ac5f8671eb6eR47

Not sure how we would write the run_event_daemon otherwise? Maybe with a typing.Protocol but that would mean it would work with anything that has a remote property even if it isn't a BaseEvent.

Maybe HasRemoteEvent turns into an abstract class with an abstract property remote?

Unfortunately mypy goes absolutely crazy with abstract properties. Even with the suggest solution of using type: ignore I couldn't get it to work. python/mypy#1362

So, I've gone back to the non-abstract version and just properly using super().__init__(remote) everywhere.

Hm, we do this in a number of places. What's the difference here?

eg~

trinity/trinity/extensibility/plugin.py

Lines 80 to 83 in 675c583

@property

@abstractmethod

def event_bus(self) -> TrinityEventBusEndpoint:

pass

Oh yeah, I was doing something stupid. Updated it. It does feel a bit boilerplatey but could be worse...

carver · 2019-05-20T20:50:00Z

trinity/protocol/common/peer_pool_event_bus.py

-        await self.cancel_token.wait()
+        self.run_daemon_task(self.handle_native_peer_messages())
+
+        await self.cancellation()


trinity/protocol/common/peer_pool_event_bus.py

trinity/protocol/eth/events.py

trinity/protocol/eth/servers.py

trinity/protocol/les/servers.py

cburgdorf · 2019-05-21T12:57:17Z

Just in case anyone wants to review this, hold back, I'm testing some things out to fine tune this.

pipermerriam

Still a somewhat incomplete review but got through most of it.

tests/conftest.py

pipermerriam · 2019-05-21T13:10:33Z

tests/core/integration_test_helpers.py

+    return event_server
+
+
+async def run_request_server(event_bus, chaindb, server_type=None):


This seems like a good candidate for an asynccontextmanager to manage setup and teardown? Not sure if that helps. Maybe teardown happens fine but in the trio world we'll have to convert it to use that pattern or provide this function with a handle to a nursery...

Sounds good in general. I just wonder how to deal with the rightward drift when you need multiple of them e.g.

async with run_event_server(): async with run_request_server(): async with run_foo(): pass

It would be nice to be able to say:

async with ( run_event_server(), run_request_server(), run_foo() ): pass

Looking for something like this? https://stackoverflow.com/a/1073814/8412986

Planning to convert this to the asynccontextmanager in this PR?

Woah, I had no idea this was actually possible! Thanks @carver for bringing it up!

This works just fine.

import asyncio from contextlib import asynccontextmanager @asynccontextmanager async def foo(): try: print("foo") yield 1 finally: print("cleaning foo") @asynccontextmanager async def bar(): try: print("bar") yield 2 finally: print("cleaning bar") async def meh(): async with foo(), bar(): print("wohoo") asyncio.run(meh())

pipermerriam · 2019-05-21T13:11:22Z

tests/core/json-rpc/test_ipc.py

@@ -87,6 +87,8 @@ def can_decode_json(potential):
        request_msg,
        event_loop):

+    # Give the event subscriptions a moment to propagate
+    await asyncio.sleep(0.01)


We've got some wait_for_XX_subscripiton APIs coming down the pipe.

trinity/plugins/builtin/request_server/plugin.py

pipermerriam · 2019-05-21T13:19:22Z

trinity/protocol/common/events.py

@@ -41,11 +54,24 @@ def expected_response_type() -> Type[PeerCountResponse]:
        return PeerCountResponse


-class DisconnectPeerEvent(BaseEvent):
+class DisconnectPeerEvent(HasRemoteEvent):
    """
    Event broadcasted when we want to disconnect from a peer
    """

    def __init__(self, remote: Node, reason: DisconnectReason) -> None:
        self.remote = remote


I'd advocate for dropping the HasRemoteEvent and just doing it explicitely. It doesn't appear to save any lines of code and I'd argue it makes readability worse.

def __init__(self, remote: Node, other_arg: int): super().__init__(remote) self.other_arg = other_arg # vs def __init__(self, remote: Node, other_arg: int): self.remote = remote self.other_arg = other_arg

trinity/protocol/common/peer_pool_event_bus.py

pipermerriam · 2019-05-21T17:54:48Z

(sorry for the too-soon review. I just saw your request to hold off)

cburgdorf · 2019-05-22T16:07:50Z

setup.py

@@ -29,7 +29,8 @@
        "plyvel==1.0.5",
        PYEVM_DEPENDENCY,
        "web3==4.4.1",
-        "lahja==0.12.0",
+        "lahja@git+https://github.com/ethereum/lahja.git@22396a8fe79809357438e917e66a7101e6e3ac01",  # noqa: E501


@carver @pipermerriam I wonder what's the best thing to do here. Ideally I'd like to get this change in sooner than later because it touches many files and especially the eth2 code changes on a such rapid pace (rightfully so 👏 ) that I have to resolve conflicts with every rebase.

So I could rebase lahja from master. However, I feel that it would just be a matter of some more days until we want to cut a new release soon after. So, if you agree that this PR could land soon, do you think it would be acceptable to keep lahja pinned to this git ref from current lahja/master just for some days?

Yeah, I think it's okay temporarily. As long as getting lahja released with the needed changes becomes top priority :). Don't want it to hold up the next trinity release.

Don't know what the plan is here but in case this got dropped a friendly reminder that master is still using this git revision.

carver · 2019-05-22T17:58:10Z

setup.py

@@ -29,7 +29,8 @@
        "plyvel==1.0.5",
        PYEVM_DEPENDENCY,
        "web3==4.4.1",
-        "lahja==0.12.0",
+        "lahja@git+https://github.com/ethereum/lahja.git@22396a8fe79809357438e917e66a7101e6e3ac01",  # noqa: E501
+        # "lahja==0.12.0",


Let's just drop this comment. We'll know which version to bring back after releasing the new lahja. Also, I think it would be good practice to use >=0.12.0<0.13.0, so that we can make bugfix patch releases on lahja without forcing a trinity patch release.

carver · 2019-05-22T18:01:05Z

tests/core/integration_test_helpers.py

-    await peer_pool_event_bus_request_handler.events.started.wait()
+    asyncio.ensure_future(event_server.run())
+
+    await event_server.events.started.wait()


Note to self: These two lines are so common that we probably want a one-liner, like await event_server.launch() that starts the BaseService in the background, but doesn't return until it's initialized. Piper's got bigger ideas for the service refactor, so we'll probably delay that syntactic sugar until after a refactor.

carver · 2019-05-22T18:02:35Z

tests/core/integration_test_helpers.py

+    return event_server
+
+
+async def run_request_server(event_bus, chaindb, server_type=None):


Looking for something like this? https://stackoverflow.com/a/1073814/8412986

carver · 2019-05-22T18:03:06Z

tests/core/integration_test_helpers.py

+    return event_server
+
+
+async def run_request_server(event_bus, chaindb, server_type=None):


Planning to convert this to the asynccontextmanager in this PR?

carver · 2019-05-22T18:05:29Z

tests/core/json-rpc/test_ipc.py

@@ -87,6 +87,8 @@ def can_decode_json(potential):
        request_msg,
        event_loop):

+    # Give the event subscriptions a moment to propagate
+    await asyncio.sleep(0.01)


Maybe add a comment with the proposed replacement code, to be used when the new API is merged/released?

trinity/protocol/common/servers.py

carver · 2019-05-22T18:30:09Z

trinity/protocol/common/servers.py

+            await self._handle_msg(remote, cmd, msg)
+        except OperationCancelled:
+            # Silently swallow OperationCancelled exceptions because otherwise they'll be caught
+            # by the except below and treated as unexpected.


Maybe add a debug log just so we notice if this is happening unexpectedly.

Wait? When would that be unexpected and how would I recognize that? Also notice, that this code did not diverge from master.

trinity/trinity/protocol/common/servers.py

Lines 58 to 70 in 675c583

async def _quiet_handle_msg(

self,

peer: BasePeer,

cmd: protocol.Command,

msg: protocol._DecodedMsgType) -> None:

try:

await self._handle_msg(peer, cmd, msg)

except OperationCancelled:

# Silently swallow OperationCancelled exceptions because otherwise they'll be caught

# by the except below and treated as unexpected.

pass

except Exception:

self.logger.exception("Unexpected error when processing msg from %s", peer)

Yeah, I guess it's fine for now. It will be handled differently after switching to trio eventually, so we can handle it then. (To answer your question, we wouldn't recognize that anything was wrong in the code. We might recognize it by browsing the logs, and noticing an unusual number or location of the logs.)

carver · 2019-05-22T18:32:00Z

trinity/protocol/eth/events.py

+    pass
+
+
+class GetNodeDataEvent(PeerPoolMessageEvent):


cc to make sure @lithp notices that the firehose request server will probably need some changes after this merges. (Like adding a new event)

Thanks for the ping! I think what you're saying is that I should race to merge Firehose before this PR lands ;)

Hahaha! This PR is like a black whole. Eventually it sucks in the entire code base...at this point my whole job will be about rebasing this PR until eternity 😅

carver · 2019-05-22T18:33:24Z

trinity/protocol/eth/peer.py

+    """
+    A ``ETHPeer`` that can be used from any process as a drop-in replacement for the actual
+    peer that sits in the peer pool. Any action performed on the ``ETHProxyPeer`` is delegated
+    to the actual peer in the pool.


Is this comment correct? There are a lot of methods on a peer. What about disconnect, for example? I only see sub_proto implemented.

That comment is a bit ahead of time haha. In the follow up PR that also touches the sync, this becomes a proper BaseService derived class and I also had a typing_extensions.Protocol ETHPeerLike that abstracted away the usage of the actual peer or the proxy peer.

I will adjust the comment for this PR so that it properly reflects the status quo.

carver · 2019-05-22T18:44:03Z

trinity/protocol/les/servers.py

        else:
-            self.logger.debug("%s msg from %s not implemented", cmd, peer)
+            self.logger.debug("%s msg not handled yet, needs to be implemented", cmd)


The peer info could theoretically be interesting (here and the eth server). If some peer is requesting unexpected/non-standard commands, it might be nice to identify the type of client that's running by scanning the logs for the peer hash and its related corresponding client string.

Notice that the only reason we could run into this, is when we mistakenly add a subscription for an event type that we then forget to handle. In other words, if we see something here, it is always our own misconfiguration.

cburgdorf · 2019-05-22T21:37:26Z

Ok, still missing some async with...on it

carver

Looks like you are debugging something, but it seems good to go. Feel free to re-ping me if the solution you end up with needs review, too.

tests/core/p2p-proto/test_server.py

cburgdorf · 2019-05-22T23:07:03Z

Looks like you are debugging something, but it seems good to go. Feel free to re-ping me if the solution you end up with needs review, too.

I think, the test that scans the shutdown for errors (which is currently a xfail) got more flaky because, somewhat ironically, the shutdown became more stable with the upgrade to a newer lahja release.

I marked the test as skip for now just because I really want to get this in now and it is 1 AM here. I swear it's the first thing I pickup tomorrow morning to resolve it properly 🙏

pipermerriam · 2019-05-23T21:01:45Z

trinity/protocol/eth/events.py

+        self._remote = remote
+        self.nodes = nodes
+
+    @property


I think that you can define these classes without inheriting from HasRemoteEvent and set this property normally and still get away with telling mypy that something takes a TypeVar('Something', bound=HasRemoteEvent) and it will still just work)

Only if HasRemoteEvent were a typing_extensions.Protocol so that structural typing is used. But that would allow any kind of thing with a remote property to be used here and not enforce proper BaseEvent checking. Unfortunately mypy doesn't allow multiple bounds (yet?) so that we can not do something like TypeVar('something', bound=[Intersection[BaseEvent, SupportsRemote]])

pipermerriam · 2019-05-23T21:02:53Z

trinity/protocol/eth/peer.py

+        )
+
+        if isinstance(cmd, ignored_commands):
+            pass


Maybe a log message at debug level so this doesn't get completely lost?

Notice that this code just moved, we've had this construct before. But I will send a follow up PR to just forward these two events because we will need that anyway when we start moving the transaction pool into a BaseIsolatedPlugin as well.

cburgdorf force-pushed the christoph/feat/isolated-request-server branch 9 times, most recently from 855e8e8 to 4f4e995 Compare May 17, 2019 11:35

pipermerriam mentioned this pull request May 17, 2019

Endpoint or Event Bus? ethereum/lahja#62

Open

pipermerriam reviewed May 17, 2019

View reviewed changes

cburgdorf force-pushed the christoph/feat/isolated-request-server branch 12 times, most recently from 74474c4 to e52dd53 Compare May 20, 2019 13:46

cburgdorf requested a review from pipermerriam May 20, 2019 13:48

carver reviewed May 20, 2019

View reviewed changes

cburgdorf force-pushed the christoph/feat/isolated-request-server branch from c6e8303 to 95a9ad3 Compare May 21, 2019 13:27

pipermerriam reviewed May 21, 2019

View reviewed changes

cburgdorf force-pushed the christoph/feat/isolated-request-server branch from 95a9ad3 to ee44f6d Compare May 22, 2019 13:29

Upgrade to new Lahja version

633c514

cburgdorf force-pushed the christoph/feat/isolated-request-server branch 3 times, most recently from d03d603 to 5c6ed9e Compare May 22, 2019 16:00

cburgdorf commented May 22, 2019

View reviewed changes

cburgdorf force-pushed the christoph/feat/isolated-request-server branch from 5c6ed9e to c231dbf Compare May 22, 2019 16:26

cburgdorf requested a review from carver May 22, 2019 16:37

carver reviewed May 22, 2019

View reviewed changes

cburgdorf added 2 commits May 22, 2019 22:50

Give the shutdown error scan more time

3840a74

Ensure each tests uses an isolated event bus

82d2334

cburgdorf force-pushed the christoph/feat/isolated-request-server branch 2 times, most recently from ba2cce3 to 7695de5 Compare May 22, 2019 21:08

lithp mentioned this pull request May 22, 2019

Add Firehose protocol #610

Closed

20 tasks

cburgdorf force-pushed the christoph/feat/isolated-request-server branch 2 times, most recently from 161e079 to aff2f7d Compare May 22, 2019 21:56

carver approved these changes May 22, 2019

View reviewed changes

tests/core/p2p-proto/test_server.py Show resolved Hide resolved

cburgdorf added 2 commits May 23, 2019 01:02

Move RequestServer out of networking process

7f74f37

Temporary skip shutdown error scan test

66c35b2

cburgdorf force-pushed the christoph/feat/isolated-request-server branch from aff2f7d to 66c35b2 Compare May 22, 2019 23:05

cburgdorf merged commit eb6eddc into ethereum:master May 22, 2019

pipermerriam reviewed May 23, 2019

View reviewed changes

cburgdorf mentioned this pull request Jul 9, 2019

Move handling of peer requests to a plugin (and it's own process) #84

Closed

	@property
	@abstractmethod
	def event_bus(self) -> TrinityEventBusEndpoint:
	pass

		return event_server


		async def run_request_server(event_bus, chaindb, server_type=None):

	async def _quiet_handle_msg(
	self,
	peer: BasePeer,
	cmd: protocol.Command,
	msg: protocol._DecodedMsgType) -> None:
	try:
	await self._handle_msg(peer, cmd, msg)
	except OperationCancelled:
	# Silently swallow OperationCancelled exceptions because otherwise they'll be caught
	# by the except below and treated as unexpected.
	pass
	except Exception:
	self.logger.exception("Unexpected error when processing msg from %s", peer)

Move RequestServer into isolated plugin #617

Move RequestServer into isolated plugin #617

Conversation

cburgdorf commented May 16, 2019 • edited

What was wrong?

How was it fixed?

To-Do

Cute Animal Picture

pipermerriam left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cburgdorf commented May 20, 2019

carver left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

carver May 22, 2019 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cburgdorf commented May 21, 2019

pipermerriam left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cburgdorf May 22, 2019 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cburgdorf May 22, 2019 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pipermerriam commented May 21, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cburgdorf commented May 22, 2019

carver left a comment

Choose a reason for hiding this comment

cburgdorf commented May 22, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cburgdorf commented May 16, 2019 •

edited

carver May 22, 2019 •

edited

cburgdorf May 22, 2019 •

edited

cburgdorf May 22, 2019 •

edited