Register for disconnected event notification by pubsub #340
Conversation
| await stream.reset() | ||
| # Force context switch for stream handlers to process the stream reset event we just emit | ||
| # before we cancel the stream handler tasks. | ||
| await asyncio.sleep(0.1) |
There was a problem hiding this comment.
This is a code smell
I would suggest looking into a mechanism to allow you to directly await this type of thing. For example, await stream.reset() could return an object that could then be awaited and wouldn't return until the underlying handlers have processed the event. Otherwise you're likely to continue to have to letter random sleeps into your code and I can say from experience in trinity that we don't want this.
There was a problem hiding this comment.
await stream.reset() along with await self.muxed_conn.close() on L40 will set the events for Mplex and Mplex_stream. This signals that Mplex and its Mplex_streams are shutting down. Then when stream handler try to read from stream it will catch StreamReset error and properly finishes.
So the problem is that changing the state of the stream does not guarantee that the stream handler task that's reading from the stream will detect the state change and finish right after. So in this case if we cancel tasks right after await stream.reset() the stream handler task will be cancelled instead of catching StreamReset error and properly finishes.
Maybe we can cancel other tasks and wait for stream handler task to finish?
- fix await stream close/reset - make `_handle_dead_peer` a sync function
NIC619
force-pushed
the
fix_pubsub_stream_to_disconnected_peer
branch
from
1097b41 to
b4a3724
Compare
NIC619
force-pushed
the
fix_pubsub_stream_to_disconnected_peer
branch
from
3350571 to
b8c7f0c
Compare
mhchia
left a comment
mhchia
left a comment
There was a problem hiding this comment.
Generally looks good to me. I left some comments about disconnecting.
| :param network: network the connection was opened on | ||
| :param conn: connection that was opened | ||
| """ | ||
| await self.dead_peers_queue.put(conn.muxed_conn.peer_id) |
There was a problem hiding this comment.
Can I understand handle_dead_peer_queue is called twice when a peer disconnects us?
There was a problem hiding this comment.
Yes, _handle_dead_peer might be called twice. Once when we disconnect from peer(network notify pubsub) and once when pubsub detect that the connection has closed.
3 participants
Implement
disconnected()inPubsubNotifeeso pubsub can be notified of peer who has disconnected from us and remove the peer's info from pubsub records.I believe this can close #306 (or it's already solved?).