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

Limit concurrency to push identify message to peers

Browse Source 
Signed-off-by: sukhman <sukhmansinghsaluja@gmail.com>
This commit is contained in:
sukhman
2025-05-23 00:39:17 +05:30
parent 2277d822f1
commit b7d62c0f85
1 changed files with 19 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
libp2p/identity/identify_push/identify_push.py
View File

@ -40,6 +40,7 @@ logger = logging.getLogger(__name__)
ID_PUSH = TProtocol("/ipfs/id/push/1.0.0") ID_PUSH = TProtocol("/ipfs/id/push/1.0.0")
PROTOCOL_VERSION = "ipfs/0.1.0" PROTOCOL_VERSION = "ipfs/0.1.0"
AGENT_VERSION = get_agent_version() AGENT_VERSION = get_agent_version()
LIMIT = trio.CapacityLimiter(10)
def identify_push_handler_for(host: IHost) -> StreamHandlerFn: def identify_push_handler_for(host: IHost) -> StreamHandlerFn:
@ -146,25 +147,26 @@ async def push_identify_to_peer(
True if the push was successful, False otherwise. True if the push was successful, False otherwise.
""" """
try: async with LIMIT:
# Create a new stream to the peer using the identify/push protocol try:
stream = await host.new_stream(peer_id, [ID_PUSH]) # Create a new stream to the peer using the identify/push protocol
stream = await host.new_stream(peer_id, [ID_PUSH])
# Create the identify message # Create the identify message
identify_msg = _mk_identify_protobuf(host, observed_multiaddr) identify_msg = _mk_identify_protobuf(host, observed_multiaddr)
response = identify_msg.SerializeToString() response = identify_msg.SerializeToString()
# Send the identify message # Send the identify message
await stream.write(response) await stream.write(response)
# Close the stream # Close the stream
await stream.close() await stream.close()
logger.debug("Successfully pushed identify to peer %s", peer_id) logger.debug("Successfully pushed identify to peer %s", peer_id)
return True return True
except Exception as e: except Exception as e:
logger.error("Error pushing identify to peer %s: %s", peer_id, e) logger.error("Error pushing identify to peer %s: %s", peer_id, e)
return False return False
async def push_identify_to_peers( async def push_identify_to_peers(
@ -179,13 +181,10 @@ async def push_identify_to_peers(
""" """
if peer_ids is None: if peer_ids is None:
# Get all connected peers # Get all connected peers
peer_ids = set(host.get_peerstore().peer_ids()) peer_ids = set(host.get_connected_peers())
# Push to each peer in parallel using a trio.Nursery # Push to each peer in parallel using a trio.Nursery
# TODO: Consider using a bounded nursery to limit concurrency # limiting concurrent connections to 10
# and avoid overwhelming the network. This can be done by using
# trio.open_nursery(max_concurrent=10) or similar.
# For now, we will use an unbounded nursery for simplicity.
async with trio.open_nursery() as nursery: async with trio.open_nursery() as nursery:
for peer_id in peer_ids: for peer_id in peer_ids:
nursery.start_soon(push_identify_to_peer, host, peer_id, observed_multiaddr) nursery.start_soon(push_identify_to_peer, host, peer_id, observed_multiaddr)