How to rate limit a coroutine and re-call the coroutine after the limit?

Question

I've been reading through the responses here: What's a good rate limiting algorithm?

The reply from Carlos A. Ibarra works great without asynchronous capabilities, but is there any way I can amend it to work asynchronously?

import time

def RateLimited(maxPerSecond):
    minInterval = 1.0 / float(maxPerSecond)
    def decorate(func):
        lastTimeCalled = [0.0]
        def rateLimitedFunction(*args,**kargs):
            elapsed = time.clock() - lastTimeCalled[0]
            leftToWait = minInterval - elapsed
            if leftToWait>0:
                time.sleep(leftToWait)
            ret = func(*args,**kargs)
            lastTimeCalled[0] = time.clock()
            return ret
        return rateLimitedFunction
    return decorate

@RateLimited(2)  # 2 per second at most
def PrintNumber(num):
    print num

if __name__ == "__main__":
    print "This should print 1,2,3... at about 2 per second."
    for i in range(1,100):
        PrintNumber(i)

Changing time.sleep(leftToWait) to await asyncio.sleep(leftToWait) and awaiting PrintNumber(i) works for the first instance, but none thereafter. I'm really new at Python and trying my best to obey an API's rate limit.

My implementation:

def rate_limited(max_per_second):
    min_interval = 1.0 / float(max_per_second)

    def decorate(func):
        last_time_called = [0.0]

        async def rate_limited_function(*args, **kargs):
            elapsed = time.clock() - last_time_called[0]
            left_to_wait = min_interval - elapsed
            if left_to_wait > 0:
                await asyncio.sleep(left_to_wait)
            ret = func(*args, **kargs)
            last_time_called[0] = time.clock()
            return ret
        return rate_limited_function
    return decorate


class Test:
    def __init__(self, bot):
        self.bot = bot

    @commands.command(hidden=True, pass_context=True)
    @checks.serverowner()
    async def test1(self, ctx):
        await self.print_number()

    @rate_limited(0.1)
    def print_number(self):
        print("TEST")

How does the sync/async distinction make a difference in the algorithm (as opposed to its implementation)? Which is to say -- I'd be expecting this to be a question about a specific attempt you'd made to adapt an implementation, and why it isn't working, vs a general "which algorithm is best?" (for which the answers on the other question are all perfectly good). — Charles Duffy, Jul 02 '18 at 21:41
@CharlesDuffy I suppose then my question would be more related to the third paragraph in my original question. Here's my tested implementation: https://paste.ee/p/is60Z I'm using it together with discord.py for a Discord bot. When I execute the `test1` command, it waits 10 seconds properly. If I run two `test1` commands back-to-back, they execute right after each other. — Craig, Jul 02 '18 at 21:51
so, I expect that the current implementation will work only to the extent that you don't have new requests come in while other ones are waiting (that is to say, if you let one request complete, but there isn't a token in the bucket to let the next one go out immediately, the next one should still wait). Is that an accurate summary of current behavior? — Charles Duffy, Jul 02 '18 at 22:04
@CharlesDuffy If I make a new request while another is pending, they all finish immediately after the first. Example flow with `@rate_limited(0.1)`: make a request > make another request 2 seconds later > make another request 2 seconds later > the first request finishes 6 seconds later (as it should) > the other 2 requests immediately finish thereafter (rather than waiting 10 seconds each) — Craig, Jul 02 '18 at 22:09
@abccd I want the second to be waited the 5 seconds and triggered after, not discarded. — Craig, Jul 02 '18 at 22:12
Building a reproducer that didn't require discord.py, btw, made this much easier to implement a fix for. — Charles Duffy, Jul 02 '18 at 22:32

Taku · Answer 1 · 2018-07-02T22:43:54.837

Here's a simple discord.py solution. This uses the on_command_error event to keep the command and run it forever until the cooldown get resolved, basically by waiting out the cooldown with asyncio.sleep:

bot = commands.Bot('?')

@bot.command(hidden=True, pass_context=True)
@commands.cooldown(1, 5, commands.BucketType.user)  # means "allow to be called 1 time every 5 seconds for this user, anywhere"
async def test(ctx):
    print("TEST")

@bot.event
async def on_command_error(exc, context: commands.Context):
    if isinstance(exc, commands.errors.CommandOnCooldown):
        while True:
            await asyncio.sleep(exc.retry_after)
            try:
                return await context.command.invoke(context)
            except commands.errors.CommandOnCooldown as e:
                exc = e

Example

In discord (assume prefix is ?):

0s> ?test
1s> ?test
2s> ?test

In console:

0s> TEST
5s> TEST
10s> TEST

Charles Duffy · Accepted Answer · 2018-07-02T22:31:03.777

One of the easiest things you can do here is to make your code re-poll the shared variables, and thus loop, rather than assuming that this current instance will be the next up after a single sleep:

import time, asyncio

def rate_limited(max_per_second):
    min_interval = 1.0 / float(max_per_second)
    def decorate(func):
        last_time_called = [0.0]
        async def rate_limited_function(*args, **kargs):
            elapsed = time.time() - last_time_called[0]
            left_to_wait = min_interval - elapsed
            while left_to_wait > 0:
                await asyncio.sleep(left_to_wait)
                elapsed = time.time() - last_time_called[0]
                left_to_wait = min_interval - elapsed
            ret = func(*args, **kargs)
            last_time_called[0] = time.time()
            return ret
        return rate_limited_function
    return decorate

@rate_limited(0.2)
def print_number():
    print("Actually called at time: %r" % (time.time(),))

loop = asyncio.get_event_loop()
asyncio.ensure_future(print_number())
asyncio.ensure_future(print_number())
asyncio.ensure_future(print_number())
asyncio.ensure_future(print_number())
loop.run_forever()

...properly emits:

Actually called at time: 1530570623.868958
Actually called at time: 1530570628.873996
Actually called at time: 1530570633.876241
Actually called at time: 1530570638.879455

...showing 5 seconds between calls (0.2 per second).

Thanks for all the help with this. So I have api.py, which contains this (which you've helped with): https://paste.ee/p/eMbR0 It should only be able to operate once every 10 seconds, correct? I set up a test command in test.py: https://paste.ee/p/o3or7 When I run the `test` command, it prints "!" instantly. Am I not implementing this properly? — Craig, Jul 02 '18 at 23:05
I'd want to see a standalone reproducer I can run to see the problem myself before commenting. (That said, heading out soon, so it might be a while before further response regardless). — Charles Duffy, Jul 02 '18 at 23:12

How to rate limit a coroutine and re-call the coroutine after the limit?

2 Answers2

Example