varun-r-mallya/py-libp2p
1
0
Fork 0
mirror of https://github.com/varun-r-mallya/py-libp2p.git synced 2026-02-12 16:10:57 +00:00
Code Issues Packages Projects Releases Wiki Activity

Catch expections in PubsubNotifee

Browse Source 
Also, add lock to avoid resource race condition
This commit is contained in:
mhchia
2019-12-24 18:31:39 +08:00
parent 2287dc95be
commit 573c049d0f
1 changed files with 25 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
libp2p/pubsub/pubsub_notifee.py
View File

@ -16,6 +16,7 @@ class PubsubNotifee(INotifee):
initiator_peers_queue: "trio.MemorySendChannel[ID]" initiator_peers_queue: "trio.MemorySendChannel[ID]"
dead_peers_queue: "trio.MemorySendChannel[ID]" dead_peers_queue: "trio.MemorySendChannel[ID]"
dead_peers_queue_lock: trio.Lock
def __init__( def __init__(
self, self,
@ -29,7 +30,9 @@ class PubsubNotifee(INotifee):
can process dead peers after we disconnect from each other can process dead peers after we disconnect from each other
""" """
self.initiator_peers_queue = initiator_peers_queue 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 = dead_peers_queue
self.dead_peers_queue_lock: trio.Lock()
async def opened_stream(self, network: INetwork, stream: INetStream) -> None: async def opened_stream(self, network: INetwork, stream: INetStream) -> None:
pass pass
@ -46,12 +49,17 @@ class PubsubNotifee(INotifee):
:param network: network the connection was opened on :param network: network the connection was opened on
:param conn: connection that was opened :param conn: connection that was opened
""" """
try: async with self.initiator_peers_queue_lock:
await self.initiator_peers_queue.send(conn.muxed_conn.peer_id) try:
except trio.BrokenResourceError: await self.initiator_peers_queue.send(conn.muxed_conn.peer_id)
# Raised when the receive channel is closed. except (
# TODO: Do something with loggers? trio.BrokenResourceError,
... trio.ClosedResourceError,
trio.BusyResourceError,
):
# Raised when the receive channel is closed.
# TODO: Do something with loggers?
...
async def disconnected(self, network: INetwork, conn: INetConn) -> None: async def disconnected(self, network: INetwork, conn: INetConn) -> None:
""" """
@ -61,7 +69,17 @@ class PubsubNotifee(INotifee):
:param network: network the connection was opened on :param network: network the connection was opened on
:param conn: connection that was opened :param conn: connection that was opened
""" """
await self.dead_peers_queue.send(conn.muxed_conn.peer_id) 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?
...
async def listen(self, network: INetwork, multiaddr: Multiaddr) -> None: async def listen(self, network: INetwork, multiaddr: Multiaddr) -> None:
pass pass