Unable to get max_delay parameter working

[ranjanprasad1996]

I am trying to achieve the following behaviour with the rate limiter:

1. The function calls should be rate limited per minute (3/ minute).
2. The overflowing requests should be in the rate limited bucket only for a maximum of 70 seconds after which it should raise an exception.

I am trying to do this with the max_delay parameter but does not seem to throw any exception after 70 seconds. Not sure if I am missing something.

import time
from multiprocessing.dummy import Pool

from pyrate_limiter import InMemoryBucket
from pyrate_limiter import Limiter, Duration, Rate

limit_s = 3
max_delay_s=70

rates = [Rate(limit_s, Duration.SECOND * 60)]

bucket = InMemoryBucket(rates)

rate_limiter = Limiter(bucket, max_delay=max_delay_s*Duration.SECOND, raise_when_fail=True).as_decorator()


def mapping(*args, **kwargs):
    return "check", 1

@rate_limiter(mapping)
def check_rate_limiter(i):
    print(f"Calling {i} {time.time()}")
    time.sleep(100)
    print(f"Done {i}")


if __name__ == "__main__":
    x = list(range(10))

    with Pool(processes=10) as p:
        p.map(check_rate_limiter, x) 

Also, for any value lower than the limit duration, it throws the
warning Required delay too large: actual=60050, expected=60000 . Any ideas on this?

[ranjanprasad1996]

@JWCook Could you please help me with this?

[paultiq]

The behavior you're seeing is explained by the use of a synchronous limiter and thread pools (multiprocessing.dummy uses threads), plus the fact that the Limiter doesn't calculate delay until after a lock is acquired.

When try_acquire is called, the task takes an RLock. This RLock means that only one task will hit the Limiter's internal "delay" at a time

PyrateLimiter/pyrate_limiter/limiter.py

Line 269 in bffe872

with self.lock:

At start:

• Task 0 gets the lock, and "acquires" a slot.
• Then, Task 1 gets the lock, and "acquires" a slot.
• Then, Task 2 gets the lock, and "acquires" a slot.
• Then, Task 3 gets the lock and blocks waiting for a slot.
• Tasks 4-9 are blocked waiting on the lock / before entering the limiter logic.
• At this point, the delay enforcement only applies to Task 3 and hasn't yet started for 4-9.

The delay is is calculated after the Task gets the lock, not before, and only accounts for the time after acquiring the lock, not before.

PyrateLimiter/pyrate_limiter/limiter.py

Line 162 in bffe872

delay = bucket.waiting(item)

I'd say use an async limiter (RedisBucket) or one for multiprocessing (PostgresBucket or the SQLiteBucket with use_file_lock).

Possible Improvements

Idea 1: Instead of measuring delay after the Lock, the start time could be capture before the lock so that total delay could be calculated. Along with this: A timeout could be passed to RLock, so that no thread waits longer than the max_delay for a lock.

Idea 2: Simplify the logic by getting rid of max_delay. Instead, add an optional timeout to RLock so that tasks can't wait more than X seconds, and leave the rest to the application to manage.

Modified test case with # of seconds

import time
from multiprocessing.dummy import Pool

from pyrate_limiter import InMemoryBucket
from pyrate_limiter import Limiter, Duration, Rate

limit_s = 3
max_delay_s=70

rates = [Rate(limit_s, Duration.SECOND * 60)]

bucket = InMemoryBucket(rates)

rate_limiter = Limiter(bucket, max_delay=max_delay_s*Duration.SECOND, raise_when_fail=True).as_decorator()


start = time.time()

def mapping(*args, **kwargs):
    return "check", 1

@rate_limiter(mapping)
def check_rate_limiter(i):
    print(f"Calling {i} {round(time.time() - start)}")
    time.sleep(100)
    print(f"Done {i} at {round(time.time() - start)}")


if __name__ == "__main__":
    x = list(range(10))

    with Pool(processes=10) as p:
        p.map(check_rate_limiter, x)