Как запустить выполнение асинхронной функции с определенной частотой выполнения?

Question

[Ninzalo]

Здравствуйте,
у меня есть код, который получает сообщение из мессенжера.
Допустим, что функция replier() отвечает на полученное сообщение. API, к которой я обращаюсь для ответа на сообщение имеет ограничение по частоте вызовов в секунду.

import asyncio
from datetime import datetime
import pickle

from lib.bot.converters import dataclass_from_dict
from lib.bot.struct import Message_struct
from config import LOCAL_IP, LOCAL_PORT, MESSAGE_REPLY_RATE


async def handle_echo(reader, writer) -> None:
    tasks = []
    while True:
        data = await reader.read()
        if not data:
            break
        message = pickle.loads(data)
        message_cls = dataclass_from_dict(
            struct=Message_struct, dictionary=message
        )
        task = asyncio.create_task(replier(message_cls=message_cls))
        tasks.append(task)

    await asyncio.gather(*tasks)
    addr = writer.get_extra_info("peername")
    print(f"Received {message_cls!r} from {addr!r}")
    writer.close()


async def replier(message_cls: Message_struct):
    # process the received data here
    print(message_cls.text, datetime.now())


async def main() -> None:
    server = await asyncio.start_server(handle_echo, LOCAL_IP, LOCAL_PORT)

    addrs = ", ".join(str(sock.getsockname()) for sock in server.sockets)
    print(f"Serving on {addrs}")

    async with server:
        await server.serve_forever()


asyncio.run(main())

Все, что я нашел в интернете - добавить asyncio.sleep(1.0 / MESSAGE_REPLY_RATE) в конце функции replier(). Но это не работает. Если мне поступает несколько сообщений из мессенжера, то функция replier() вызывается одновременно
Как добавить частоту ответа на сообщения?

Answer 1

[Vindicar]

Храни в asyncio.Queue очередь запросов к API. Отдельная задача пусть выбирает запросы из очереди, отправляет, получает ответ и оповещает о результате. Например, так.

import asyncio
import typing

class ThrottledResource:
    def __init__(self, delay: float):
        self._delay = delay
        self._queue = asyncio.Queue()
        self._task = None
    
    def start(self):
        self._task = asyncio.create_task(self._work_loop)
    
    def stop(self):
        self._task.cancel()
        self._task = None

    # этот метод вызывается клиентским кодом, получает параметры и возвращает отклик спустя время.
    async def query(self, params):
        future = asyncio.Future()  # Future просигналит, когда наш запрос будет обслужен
        await self._queue.put((future, params))
        result = await future  # корутина спит, пока запрос не обслужат
        return result

    async def _work_loop(self):
        while True:
            future, params = await. self._queue.get()  # ждем, пока не придёт запрос
            try:
                result = await call_api(params)  # тут делаем асинхронное обращение к сервису
            except Exception as err:
                future.set_exception(err)  # была ошибка - теперь await future выкинет исключение
            else:
                future.set_result(result)  # полуен результат - await future вернёт его
            self._queue.task_done()  # каждому успешному get() соответствует task_done()
            asyncio.sleep(self._delay)  # можно учесть, сколько времени делался запрос. Но стоит ли?

Код примерный, но идею передаёт. Использоваться будет как-то так

api = ThrottledResource(delay=1.0)
api.start()
...
result = await api.query(params)  # await подождёт, пока не дойдёт очередь до нашего запроса

Нужно добавить обработку ошибок, корректное завершение работы при наличии задач в очереди, и так далее.
Технически вместо класса можно было реализовать это всё в виде декоратора над replier(), но это уже на вкус и цвет.

Comments

[Ninzalo]

Вот финальный код, включая мой пример:

Код

import asyncio
from datetime import datetime
import pickle

from lib.bot.converters import dataclass_from_dict
from lib.bot.struct import Message_struct
from config import LOCAL_IP, LOCAL_PORT, MESSAGE_REPLY_RATE

class ThrottledResource:
    def __init__(self, delay: float):
        self._delay = delay
        self._queue = asyncio.Queue()
        self._task = None

    def start(self):
        self._task = asyncio.create_task(self._work_loop())
    
    def stop(self):
        self._task.cancel()
        self._task = None

    # этот метод вызывается клиентским кодом, получает параметры и возвращает отклик спустя время.
    async def query(self, params):
        future = asyncio.Future()  # Future просигналит, когда наш запрос будет обслужен
        await self._queue.put((future, params))
        result = await future  # корутина спит, пока запрос не обслужат
        return result

    async def _work_loop(self):
        while True:
            future, params = await self._queue.get()  # ждем, пока не придёт запрос
            try:
                result = await replier(params)  # тут делаем асинхронное обращение к сервису
            except Exception as err:
                future.set_exception(err)  # была ошибка - теперь await future выкинет исключение
            else:
                future.set_result(result)  # полуен результат - await future вернёт его
            self._queue.task_done()  # каждому успешному get() соответствует task_done()
            await asyncio.sleep(self._delay)


async def handle_echo(reader, writer, message_throttler) -> None:
    tasks = []
    while True:
        data = await reader.read()
        if not data:
            break
        message = pickle.loads(data)
        message_cls = dataclass_from_dict(
            struct=Message_struct, dictionary=message
        )
        
        task = asyncio.create_task(message_throttler.query(message_cls))
        tasks.append(task)

    await asyncio.gather(*tasks)
    addr = writer.get_extra_info("peername")
    # print(f"Received {message_cls!r} from {addr!r}")
    writer.close()


async def replier(message_cls: Message_struct):
    # process the received data here
    print(message_cls.text, datetime.now())


async def main() -> None:
    message_throttler = ThrottledResource(delay=1.0/MESSAGE_REPLY_RATE)
    message_throttler.start()
    server = await asyncio.start_server(
        lambda reader, writer: handle_echo(reader, writer, message_throttler),
        LOCAL_IP,
        LOCAL_PORT
    )

    addrs = ", ".join(str(sock.getsockname()) for sock in server.sockets)
    print(f"Serving on {addrs}")

    async with server:
        await server.serve_forever()


asyncio.run(main())

[Ninzalo]

Vindicar Но что делать, если, допустим, получение ответа от api занимает 5 секунд?
В вашем примере в итоге получается, что запросы идут последовательно (ответ получил -> новый запрос отправил и тд).
Есть ли способ все равно делать эти запросы асинхронно, но с определенной частотой?
Или же я неправильно использовал ваш пример?

[Vindicar]

Ninzalo, ну тут уже зависит от того, как api понимает "интервал между запросами".
Я бы страховался и делал интервал от конца запроса до начала следующего.
Но если от начала до начала, то можно выкрутиться, наверное... вместо await в цикле делать create_task(). Всё равно сигнализация делается через future, так что можно задачу не await'ить.

#
    async def _single_response(self, params, future):
        try:
            result = await replier(params)  # тут делаем асинхронное обращение к сервису
        except Exception as err:
            future.set_exception(err)  # была ошибка - теперь await future выкинет исключение
        else:
            future.set_result(result)  # полуен результат - await future вернёт его

    async def _work_loop(self):
        while True:
            future, params = await self._queue.get()  # ждем, пока не придёт запрос
            asyincio.create_task(self._single_response(params, future))
            self._queue.task_done()  # каждому успешному get() соответствует task_done()
            await asyncio.sleep(self._delay)

[Ninzalo]

Vindicar, спасибо огромное, очень сильно помогли, все прекрасно работает

[Vindicar]

Ninzalo, я интереса ради переделал код на декоратор и добавил фичу для автоматического запуска/остановки _work_loop().

Если интересно

import typing
import asyncio
import functools


def throttled(delay: float, measure: typing.Literal['end_to_start', 'start_to_start'] = 'start_to_start'):
    def decorator(actual_func: typing.Coroutine) -> typing.Coroutine:
        queue = None
        
        async def _single_query(future, args, kwargs):
            try:
                result = await actual_func(*args, **kwargs)  # тут делаем асинхронное обращение к сервису
            except Exception as err:
                future.set_exception(err)  # была ошибка - теперь await future выкинет исключение
            else:
                future.set_result(result)  # полуен результат - await future вернёт его
        
        async def _work_loop():
            nonlocal queue
            while True:
                try:
                    # ждем, пока не придёт запрос, или пока не закончится таймаут
                    future, args, kwargs = await asyncio.wait_for(queue.get(), delay)
                except asyncio.TimeoutError:  # новые запросы долго не приходят, сворачиваем работу, чтобы не тратить ресурсы
                    queue = None
                    return
                task = _single_query(future, args, kwargs)
                if measure == 'start_to_start':
                    asyncio.create_task(task)
                else:
                    await task
                queue.task_done()  # каждому успешному get() соответствует task_done()
                await asyncio.sleep(delay)
        
        @functools.wraps(actual_func)
        async def query(*args, **kwargs):
            nonlocal queue  # обращение к переменной выше уровнем, но не глобальной
            future = asyncio.Future()  # Future просигналит, когда наш запрос будет обслужен
            if queue is None:  # либо это первый запрос, либо запросы долго не приходили, и мы свернули работу
                queue = asyncio.Queue()
                asyncio.create_task(_work_loop())
            await queue.put((future, args, kwargs))
            return await future
        
        return query
    
    return decorator

Использование:

@throttled(delay=1.0/MESSAGE_REPLY_RATE, measure='start_to_start')
async def replier(message_cls: Message_struct):
    ...

[Ninzalo]

Vindicar, протестировал ваш код. После небольшого рефакторинга все заработало.
Отличное решение! Теперь не приходится создавать этот класс и тащить его через весь код
Спасибо!

Код

import asyncio
from datetime import datetime
import pickle

from lib.bot.converters import dataclass_from_dict
from lib.bot.struct import Message_struct
from config import LOCAL_IP, LOCAL_PORT, MESSAGE_REPLY_RATE

import typing
import functools


def throttled(
    delay: float,
    measure: typing.Literal[
        "end_to_start", "start_to_start"
    ] = "start_to_start",
):
    def decorator(actual_func: typing.Coroutine) -> typing.Coroutine:
        queue = None

        async def _single_query(future, args, kwargs):
            try:
                result = await actual_func(
                    *args, **kwargs
                )  # тут делаем асинхронное обращение к сервису
            except Exception as err:
                future.set_exception(
                    err
                )  # была ошибка - теперь await future выкинет исключение
            else:
                future.set_result(
                    result
                )  # полуен результат - await future вернёт его

        async def _work_loop():
            nonlocal queue
            while True:
                try:
                    # ждем, пока не придёт запрос, или пока не закончится таймаут
                    future, args, kwargs = await asyncio.wait_for(
                        queue.get(), delay
                    )
                except (
                    asyncio.TimeoutError
                ):  # новые запросы долго не приходят, сворачиваем работу, чтобы не тратить ресурсы
                    queue = None
                    return
                task = _single_query(future, args, kwargs)
                if measure == "start_to_start":
                    asyncio.create_task(task)
                else:
                    await task
                queue.task_done()  # каждому успешному get() соответствует task_done()
                await asyncio.sleep(
                    delay
                )  # можно учесть, сколько времени делался запрос. Но стоит ли?

        @functools.wraps(actual_func)
        async def query(*args, **kwargs):
            nonlocal queue  # обращение к переменной выше уровнем, но не глобальной
            future = (
                asyncio.Future()
            )  # Future просигналит, когда наш запрос будет обслужен
            if (
                queue is None
            ):  # либо это первый запрос, либо запросы долго не приходили, и мы свернули работу
                queue = asyncio.Queue()
                asyncio.create_task(_work_loop())
            await queue.put((future, args, kwargs))
            return await future
        return query
    return decorator

async def handle_echo(reader, writer) -> None:
    tasks = []
    while True:
        data = await reader.read()
        if not data:
            break
        message = pickle.loads(data)
        message_cls = dataclass_from_dict(
            struct=Message_struct, dictionary=message
        )
        task = asyncio.create_task(replier(message_cls))
        tasks.append(task)
    addr = writer.get_extra_info("peername")
    print(f"Received {message_cls!r} from {addr!r}")
    await asyncio.gather(*tasks)
    writer.close()


@throttled(delay=1.0/MESSAGE_REPLY_RATE, measure='start_to_start')
async def replier(message_cls: Message_struct):
    # send answer to user
    print(
        f"Answered to {message_cls.user_id}, with text "
        f"'{message_cls.text}' at {datetime.now()}"
    )


async def main() -> None:
    server = await asyncio.start_server(
        lambda reader, writer: handle_echo(reader, writer),
        LOCAL_IP,
        LOCAL_PORT,
    )
    addrs = ", ".join(str(sock.getsockname()) for sock in server.sockets)
    print(f"Serving on {addrs}")
    async with server:
        await server.serve_forever()
asyncio.run(main())

Answer 2

[rPman]

await asyncio.sleep(1.0 / MESSAGE_REPLY_RATE)
нужно добавлять в тело цикла, после tasks.append(task), добавив await

тебе ведь нужно не создавать новые task чаще чем указанная частота, вот и создавай их с интервалом.

Comments

[Ninzalo]

При отправке, допустим, 10 сообщений из мессенжера, они все равно обработаются практически одновременно (учитывая мой пример, где обработка = вывод message.text и datetime.now())

[rPman]

тебе нужно более строгое следование ограничения по частоте запросов и при этом максимально оперативный вызов метода?

для начала нужно понять, какой алгоритм используется сервером для ограничения лимита (есть вариант когда сервер запоминает до N последних времен запросов по дате либо вариант, когда лимит установлен на определенный фиксированный интервал времени, например каждые N секунд) и в зависимости от него симулировать этот расчет лимита времени и только при его возникновении вставлять sleep

очень часто сервисы, лимитирующие запросы, в заголовках выдают информацию об остатках лимита, пользуйся этим.