Как решить подобную задачу используя async/await/asyncio?

Question

[Max Payne]

Я занимаюсь разработкой ботов для вконтакте и обычно использую библиотеку multiprocessing, но сегодня пришла в голову идея переписать часть кода с использованием asyncio, почему-то подумалось что существенно выиграю в производительности.

Прочитал мануал по нему, потом ещё один, потом ещё двадцать и ничего не понял. На сколько прост и лёгок multiprocessing, на столько же непонятен asyncio. То ли мануалы состоят более чем полностью из г*вна и палок, то ли сам asyncio.

Наверное, немного помогу описав то, как у меня всё работает сейчас:
1. Для каждого бота создаётся новый процесс

mp.Process(name=bot.name + 'bot', target=self.bot, args=(bot.name, bot.status)).start()

2. Далее в этом процессе создаётся переменная с очередью, в которую ещё один процесс складывает сообщения для отправки

def bot(self, bot_name, bot_status, MAX_DELAY=4.0, MIN_DELAY=1.6, STEP=0.2, MAX_MSG_LEN=4000, MAX_ATTS=10):
        q = mp.Queue()
        mp.Process(name=bot_name + 'bot_client', target=self.client, args=(q, bot_name, bot_status)).start()
        while True:
            MSG = q.get()
            if type(MSG) != dict: continue
        ... # далее идёт неинтересная отправка сообщения в диалог

def client(self, q, bot_name, bot_status, wait=45, lp_version=4, mode=2): # получение event'ов, обработка и добавление их в очередь на отправку
        while True:
            try:
                lps = self.bots[bot_status][bot_name].getLPS(lp_version)
                while True:
                    im = requests.get('https://' + lps['server'], params={'key': lps['key'], 'ts': lps['ts'], 'act': 'a_check', 'mode': mode, 'wait': wait, 'version': lp_version}).json()
                    if im.get('updates', None) != None:
                        lps['ts'] = im['ts']
                        if len(im['updates']) > 0:
                            for update in im['updates']:
                                if update[0] != 4: continue
                                u = self.bots[bot_status][bot_name].check_update(update, lp_version)
                                if u != None: mp.Process(target=lambda u: q.put_nowait(self.bots[bot_status][bot_name].f(u)), args=(u,)).start()
                    else:
                        print(bot_name, 'failed', im.get('failed', None))
                        if im.get('failed', None) == 1: lps['ts'] = im['ts']
                        else: lps = self.bots[bot_status][bot_name].getLPS(lp_version)
            except Exception as e: print(e)

Как решить подобную задачу используя async/await/asyncio?

Answer 1

[Dimonchik]

надо чуток перестроить мозги

но выигрыш в памятипожирании того стоит

Comments

[Max Payne]

Бесполезно, как и остальные статьи на эту тему. Прочитав код можно накопить с десяток вопросов на которых нет ответов в статье.

Answer 2

[Алексей]

В первом приближении на asyncio ваши боты будут представлять из себя корутины а очереди надо заменить на таковые из пакета asyncio. Но одним asyncio вы не обойдётесь так как он слишком низкоуровневый, вам же придётся выполнять какие то http запросы, возможно слушать запросы от vk api и т.п. Реализовывать это самому на asyncio безсмысленно, лучше взять готовые реализации вроде aiohttp как вам уже советовали. Возможно что бы разобратся вам поможет вот такой синтетический пример на голом asyncio, тут несколько корутин получают собщения из своих очередей и просто печатают их на экран:

код

import asyncio
import random


class Bot:
    def __init__(self, bot_name, queue):
        self._name = bot_name
        self._queue = queue

    async def start(self):
        try:
            while True:
                data = await self._queue.get()
                print('{}: {}'.format(self._name, data))
        except asyncio.CancelledError:
            print("{}: stoped".format(self._name))


async def message_generator(bots):
    i = 0
    while True:
        for bot in bots:
            i += 1
            await bot["queue"].put("msg {}".format(i))
            await asyncio.sleep(((random.random()/2.0)+0.1))


if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    bots = []
    n = 4
    for i in range(n):
        q = asyncio.Queue()
        t = loop.create_task(Bot("bot{}".format(i), q).start())
        bot = {
            "queue": q,
            "task": t
        }
        bots.append(bot)
    try:
        loop.run_until_complete(message_generator(bots))
    except KeyboardInterrupt:
        for bot in bots:
            bot["task"].cancel()
            loop.run_until_complete(bot["task"])
        loop.stop()

Ваши боты технически будут работать так же но вместо вывода в терминал выполнять какую то свою логику.

Comments

[Max Payne]

Боты не имеют отношения к телеграмму и работют через vk long poll, но за пример спасибо, посмотрю.

[Алексей]

да, это я перепутал. Исправил на vk :)