Combine inbound and outbound connection concepts #85
Conversation
pipermerriam
force-pushed
the
piper/combine-inbound-and-outbound-connection-concepts
branch
from
f20dfc1
to
054369b
Compare
| @@ -141,10 +187,16 @@ def __init__( | |||
| return | |||
|
|
|||
| if isinstance(message, Broadcast): | |||
| self.new_msg_func(message) | |||
| await self.new_msg_func(message) | |||
There was a problem hiding this comment.
This change is effctively changing us from using Queue.put_nowait to Queue.put which seems more appropriate (but I don't really know why). Thoughts?
There was a problem hiding this comment.
Yep, it looks like this has been a loophole where we didn't have backpressure from the pointer after we read the incoming messages to the point where we forward it to the queue. I guess turning this into await Queue.put fixes that loophole so that we have backpressure all along the way.
pipermerriam
force-pushed
the
piper/combine-inbound-and-outbound-connection-concepts
branch
from
054369b
to
62cc009
Compare
lahja/asyncio/endpoint.py
Outdated
| """ | ||
|
|
||
| logger = logging.getLogger("lahja.endpoint.InboundConnection") |
There was a problem hiding this comment.
"lahja.endpoint.InboundConnection" -> "lahja.endpoint.RemoteEndpoint"
| def __init__( | ||
| self, conn: Connection, new_msg_func: Callable[[Broadcast], None] | ||
| self, | ||
| name: Optional[str], |
There was a problem hiding this comment.
What's the reason for making this optional? I feel we should treat endpoint names as mandatory considering their importance. E.g. Answering a request can only be made efficient if the name of origin endpoint is known.
There was a problem hiding this comment.
The name is an implicit part of the connection. When a server receives an incoming connection over the socket it doesn't know the name of the endpoint on the other side of the connection. This is the concept of a half connection.
I however now realize, that with subscriptions, we can potentially get rid of the distinction between half/full connections since we still have a mechanism for determining which endpoints are acceptable to send over.
However, I am going to open an issue to deal with this separately.
| @@ -141,10 +187,16 @@ def __init__( | |||
| return | |||
|
|
|||
| if isinstance(message, Broadcast): | |||
| self.new_msg_func(message) | |||
| await self.new_msg_func(message) | |||
There was a problem hiding this comment.
Yep, it looks like this has been a loophole where we didn't have backpressure from the pointer after we read the incoming messages to the point where we forward it to the queue. I guess turning this into await Queue.put fixes that loophole so that we have backpressure all along the way.
pipermerriam
deleted the
piper/combine-inbound-and-outbound-connection-concepts
branch
|
Thanks for the ping but no complaints here! I had originally broken them into two classes because I thought that made it easier to reason about how endpoints communicate with inbound and outbound peers, but I agree that it was wasteful to use 2x the number of connections and if merging them makes trio easier I'm all in 👍 |
| return | ||
|
|
||
| # Broadcast to every connected Endpoint that is allowed to receive the event | ||
| compressed_item = self._compress_event(item) | ||
| disconnected_endpoints = [] | ||
| for name, remote in list(self._outbound_connections.items()): | ||
| for name, remote in list(self._full_connections.items()): | ||
| # list() makes a copy, so changes to _outbount_connections don't cause errors |
There was a problem hiding this comment.
This comment should be updated to reference _full_connections. My careless typo probably foiled your search-replace 😅
built on top of #70
What was wrong?
The concept op remote connections is currently divide across two classes. One for inbound and one for outbound. The reality is these connections are bi-directional, we just don't take advantage of this.
It is currently not intuitive that if I do
client.connect_to_endpoint(server)and thenserver.broadcast()that the client will not receive the broadcast.For connections that need to talk both ways, leveraging both directions of the connection should reduce the number of sockets and connections we need to make.
How was it fixed?
In the upcoming
trioimplementation in #56 there is an API that leverages the bi-directional nature of connections to makeserver.broadcast()send things back to clients that are connected.This PR is a prerequisite to that functionality, collapsing the connection concept down to a single class. This PR makes it capable of using the connection in both directions but does not actually do anything with the other side of the connection yet.
Summary of approach.
Cute Animal Picture