Add ResponseDemux #6190
Add ResponseDemux #6190
Conversation
maffoo
left a comment
maffoo
left a comment
There was a problem hiding this comment.
Minor comments, then LGTM
| Returns: | ||
| A future for the response, to be fulfilled when publish is called. | ||
| """ | ||
| response_future: asyncio.Future = asyncio.get_running_loop().create_future() |
There was a problem hiding this comment.
nit: prefer to pass in a loop to __init__ and store it as a class attribute. This makes explicit that the class should be used with a single event loop (e.g. in a single thread).
There was a problem hiding this comment.
Hmm, this object is currently not initialized inside an asyncio coroutine (see https://github.com/quantumlib/Cirq/pull/6145/files#diff-064e02971da25c80aed74456243f329557be04f837822bfbdda8dfc403e23432R144). It makes sense to me to consider limiting the scope of ResponseDemux to just the _manage_stream() coroutine, but my preference is to do that in a follow-up PR.
| for future in self._subscribers.values(): | ||
| if not future.done(): | ||
| future.set_exception(exception) | ||
| self._subscribers = {} |
There was a problem hiding this comment.
Consider call clear rather than creating a new dict:
| self._subscribers = {} | |
| self._subscribers.clear() |
| if response.message_id not in self._subscribers: | ||
| return | ||
|
|
||
| future = self._subscribers.pop(response.message_id) | ||
| if not future.done(): | ||
| future.set_result(response) |
There was a problem hiding this comment.
Can pass a default to pop to avoid a separate membership check:
| if response.message_id not in self._subscribers: | |
| return | |
| future = self._subscribers.pop(response.message_id) | |
| if not future.done(): | |
| future.set_result(response) | |
| future = self._subscribers.pop(response.message_id, None) | |
| if future and not future.done(): | |
| future.set_result(response) |
…t/response-demux-asyncio-heavy
wcourtney
left a comment
wcourtney
left a comment
There was a problem hiding this comment.
Initial comments on core logic, not yet tests. I think I need some clarification about the usage. Thanks!
| """A event demultiplexer for QuantumRunStreamResponses, as part of the async reactor pattern.""" | ||
|
|
||
| def __init__(self): | ||
| self._subscribers: Dict[str, asyncio.Future] = {} |
There was a problem hiding this comment.
Can you documentation for this field, e.g. to describe the expected use/meaning of the keys and values?
| async def test_no_subscribers_does_not_throw(self, demux): | ||
| demux.publish(RESPONSE0) | ||
|
|
||
| # expect no exceptions |
There was a problem hiding this comment.
Why not? This sounds like it would typically be an error.
There was a problem hiding this comment.
If a response comes back from the server and the caller no longer wants it, it can be dropped. This scenario could happen and shouldn't be an error condition IMO.
Updated publish() docstring to reflect this.
There was a problem hiding this comment.
My concern is that this could also represent a bug that we would suppress, e.g. if either we forget to subscribe after sending the request or if the server sends an invalid message id. Can we disambiguate the legit and erroneous cases? I don't want to de-rail; could it be left as a TODO with an issue filed?
There was a problem hiding this comment.
I think even if we caught the erroneous cases, because this is part of client code, the error message isn't actionable for the user. It would be better to catch those cases through tests, and the scenarios you described will be covered by StreamManager tests via timeouts.
wcourtney
left a comment
wcourtney
left a comment
There was a problem hiding this comment.
Overall LGTM % some thing in the cancellation/unsubscription semantics it would be nice to resolve.
| async def test_no_subscribers_does_not_throw(self, demux): | ||
| demux.publish(RESPONSE0) | ||
|
|
||
| # expect no exceptions |
There was a problem hiding this comment.
My concern is that this could also represent a bug that we would suppress, e.g. if either we forget to subscribe after sending the request or if the server sends an invalid message id. Can we disambiguate the legit and erroneous cases? I don't want to de-rail; could it be left as a TODO with an issue filed?
| del self._subscribers[message_id] | ||
| """Indicates that the caller is no longer waiting for the response matching message_id. | ||
| This helps ResponseDemux free up resources. |
There was a problem hiding this comment.
This sounds like something that the client should not be responsible for unless it's a real issue and is unresolvable without client intervention.
There was a problem hiding this comment.
Turns out this also isn't necessary after all. The only time unsubscribe is called is when the stream fails, at which point the entire ResponseDemux is cleared after publishing the exception all subscribers. Removing!
4 participants
ResponseDemux implementation, from the prototype #6145
This is just a single dict, but with restricted access patterns. This data structure hopefully makes the broader streaming client code easy to reason about and more robust, with its dedicated unit tests.
This version is different from the prototype in that the caller is expected to generate the message ID. This is because the unsubscribe method in the prototype is broken: it cannot unsubscribe with just the caller-provided request because the message ID is missing.
Open to a better class name!
@maffoo @wcourtney