StarMirror/py-libp2p
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Remove locks in PubsubNotifee

Browse Source 
- Change `open_memory_channel(0)` to `open_memory_channel(math.inf)`, to
avoid `peer_queue.send` and `dead_peer_queue.send` blocking. This allows
us to remove the locks.
- Only catch `trio.BrokenResourceError`, which is caused by Pubsub when
it's closing.
This commit is contained in:
mhchia 2020-01-27 14:30:44 +08:00
parent 92ea35e147
commit c3ba67ea87
No known key found for this signature in database
GPG Key ID: 389EFBEA1362589A
2 changed files with 19 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
libp2p/pubsub/pubsub.py
View File

@ -95,15 +95,15 @@ class Pubsub(Service, IPubsub):
# Attach this new Pubsub object to the router
self.router.attach(self)
peer_channels = trio.open_memory_channel[ID](0)
dead_peer_channels = trio.open_memory_channel[ID](0)
peer_send, peer_receive = trio.open_memory_channel[ID](math.inf)
dead_peer_send, dead_peer_receive = trio.open_memory_channel[ID](math.inf)
# Only keep the receive channels in `Pubsub`.
# Therefore, we can only close from the receive side.
self.peer_receive_channel = peer_channels[1]
self.dead_peer_receive_channel = dead_peer_channels[1]
self.peer_receive_channel = peer_receive
self.dead_peer_receive_channel = dead_peer_receive
# Register a notifee
self.host.get_network().register_notifee(
PubsubNotifee(peer_channels[0], dead_peer_channels[0])
PubsubNotifee(peer_send, dead_peer_send)
)
# Register stream handlers for each pubsub router protocol to handle

43
libp2p/pubsub/pubsub_notifee.py
View File

@ -16,7 +16,6 @@ class PubsubNotifee(INotifee):
initiator_peers_queue: "trio.MemorySendChannel[ID]"
dead_peers_queue: "trio.MemorySendChannel[ID]"
dead_peers_queue_lock: trio.Lock
def __init__(
self,
@ -30,15 +29,13 @@ class PubsubNotifee(INotifee):
can process dead peers after we disconnect from each other
"""
self.initiator_peers_queue = initiator_peers_queue
self.initiator_peers_queue_lock = trio.Lock()
self.dead_peers_queue = dead_peers_queue
self.dead_peers_queue_lock = trio.Lock()
async def opened_stream(self, network: INetwork, stream: INetStream) -> None:
pass
...
async def closed_stream(self, network: INetwork, stream: INetStream) -> None:
pass
...
async def connected(self, network: INetwork, conn: INetConn) -> None:
"""
@ -49,17 +46,11 @@ class PubsubNotifee(INotifee):
:param network: network the connection was opened on
:param conn: connection that was opened
"""
async with self.initiator_peers_queue_lock:
try:
await self.initiator_peers_queue.send(conn.muxed_conn.peer_id)
except (
trio.BrokenResourceError,
trio.ClosedResourceError,
trio.BusyResourceError,
):
# Raised when the receive channel is closed.
# TODO: Do something with loggers?
...
try:
await self.initiator_peers_queue.send(conn.muxed_conn.peer_id)
except trio.BrokenResourceError:
# The receive channel is closed by Pubsub. We should do nothing here.
...
async def disconnected(self, network: INetwork, conn: INetConn) -> None:
"""
@ -69,20 +60,14 @@ class PubsubNotifee(INotifee):
:param network: network the connection was opened on
:param conn: connection that was opened
"""
async with self.dead_peers_queue_lock:
try:
await self.dead_peers_queue.send(conn.muxed_conn.peer_id)
except (
trio.BrokenResourceError,
trio.ClosedResourceError,
trio.BusyResourceError,
):
# Raised when the receive channel is closed.
# TODO: Do something with loggers?
...
try:
await self.dead_peers_queue.send(conn.muxed_conn.peer_id)
except trio.BrokenResourceError:
# The receive channel is closed by Pubsub. We should do nothing here.
...
async def listen(self, network: INetwork, multiaddr: Multiaddr) -> None:
pass
...
async def listen_close(self, network: INetwork, multiaddr: Multiaddr) -> None:
pass
...