How do I implement throttle on async validators ?

[varun-r-mallya]

https://github.com/libp2p/py-libp2p/blob/main/libp2p/pubsub/pubsub.py#L585
I found this TODO lurking in pubsub and want to do something about it.
Can I have a reference implementation or atleast an idea on how to fix this ?
@seetadev

[seetadev]

@varun-r-mallya : Definitely, I would recommend you to look at pub-sub example first developed by @sumanjeet0012.

Wish to also recommend you to open an issue after discussion with @mystical-prog (Vraj )and @sumanjeet0012 (Sumanjeet). They will be able to guide you the best here as they are actively working on PRs focused on pub-sub and gossipsub.

I'll catch up as soon as I have completed multiple meetings with multiformats team: #645

[seetadev]

@varun-r-mallya : To implement throttling on async validators in py-libp2p, you'll want to ensure that your validator functions (often called in gossipsub for pubsub message validation) don't overwhelm the system, especially if you're receiving many messages per second.

Here’s a practical approach using asyncio (trio) semaphores to throttle concurrency and rate-limiting to control message processing rate.

1. Use a trio.Semaphore to limit concurrent validators
   This ensures no more than N messages are being validated at the same time.

2. Use a rate limiter to limit validations per second (optional)
   Use aioratelimit, trio-throttle, or your own implementation to prevent abuse or overload.

3. Register the throttled validator with gossipsub

Some notes from collaborative efforts:

• For dynamic throttling, you can modify the semaphore or rate limit at runtime based on system load.

• Always make sure your validator doesn’t block the event loop (e.g., avoid time.sleep, use await asyncio.sleep).

• Consider logging or metrics for dropped/skipped messages due to throttling.

Would like to hear feedback from @mystical-prog , @acul71 , @sumanjeet0012, @Khwahish29.

On the same note, wish to share that go-libp2p has nice rate limiter capability. Would be nice to study it too.

[varun-r-mallya]

I think I'll begin by locating similar code in go-libp2p and then moving on to reading things around it. Can you please point me to where similar code might be in the go implementation ?

[seetadev]

@varun-r-mallya : Definitely, I am been following up on go-libp2p developments. Would recommend to first visit the specs for better understanding: https://github.com/libp2p/specs/blob/master/pubsub/README.md#message-validation

In reference to the specific file, please visit https://github.com/libp2p/go-libp2p-pubsub/blob/master/validation.go (please study Validate and ValidateAsync related logic. Validators in go-libp2p are throttled via internal validator queues and worker pools.) This file will help you understand message validator concurrency and Rate Limiting.

Next, in the same file, look for validationQueue structure. This controls how many messages are being validated concurrently. According to the specs, it is controlled with a channel-based semaphore mechanism (chan struct{} as a concurrency limiter).

Please also visit some of the foundational efforts shared at libp2p/go-libp2p-pubsub#169 (interesting set of pointers by Raul). Also, visit libp2p/go-libp2p-pubsub#172 and libp2p/go-libp2p-connmgr#37

Here’s some pseudo code to model this in Python with Trio:

import trio
from trio_throttle import RateLimiter
validator_semaphore = trio.Semaphore(32)
rate_limiter = RateLimiter(rate=100)

async def throttled_validator(msg):
async with validator_semaphore:
async with rate_limiter:
# Your async validation logic here
is_valid = await validate_message(msg)
return is_valid

Hope this helps :)

[varun-r-mallya]

Thanks a lot for the info. I'll get back to working on this after I've put some effort into noise and interop as well.