Как использовать asyncio socket server с multiprocessing?

Question

[Coder 1448]

Данный код просто замирает.

Код

import asyncio
import random
from multiprocessing import Pool
from concurrent.futures import ProcessPoolExecutor

from loguru import logger

HOST = 'localhost'
PORT = random.randint(2 ** 10, 2 ** 16 - 1)


def run_async(func):
    asyncio.run(func())


async def read_all(reader: asyncio.StreamReader) -> bytes:
    data = bytes()
    while not reader.at_eof():
        data += await reader.read(1024)

    return data


async def server_handler(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
    logger.debug('server start receiving...')
    data = await read_all(reader)
    logger.info(f'server received {data!r}')

    data = b'server\'s response'
    logger.debug(f'server preparing to send {data!r}')
    writer.write(data)
    await writer.drain()
    logger.info(f'server sent {data!r}')


async def run_server():
    logger.debug('server is starting...')
    server = await asyncio.start_server(server_handler, HOST, PORT)
    logger.debug('server started')
    async with server:
        logger.debug('serving server forever')
        await server.serve_forever()


async def run_client():
    logger.debug('client is starting...')
    reader, writer = await asyncio.open_connection(HOST, PORT)
    logger.debug('client started')

    while True:
        data = b'client\'s request'
        logger.debug(f'client preparing to send {data!r}')
        writer.write(data)
        await writer.drain()
        logger.info(f'client sent {data!r}')

        logger.debug('client starting receiving...')
        data = await read_all(reader)
        logger.info(f'client received {data!r}')

        await asyncio.sleep(3)


if __name__ == '__main__':
    with Pool(processes=2) as pool:
        pool.map(run_async, [run_server, run_client])

    # также пробовал это
    # with ProcessPoolExecutor(max_workers=2) as executor:
    #     executor.map(run_async, [run_server, run_client])

При использовании библиотеки socket и вызова server.setblocking(False)всё работает.

Comments

[Дмитрий]

Пул не для таких задач предназначен

[Сергей Горностаев]

Желательно никак не использовать. Смешение разных типов конкурентности - это почти всегда ошибка и почти всегда из-за непонимания принципов.

[Владимир Олохтонов]

Слишком сильное утверждение, имхо.

Есть сценарии в которых ThreadPoolExecutor и ProcessPoolExecutor вполне можно использовать в сочетании с asyncio для распараллеливания и вынесения долгих синхронных кусков из main thread.

Для ThreadPoolExecutor это в первую очередь вызовы внешних модулей, отпускающих GIL и синхронные io-bound модули (тот же minio для s3 например).

ProcessPoolExecutor стоит использовать в основном если стоит задача распараллелить python-код.

[Владимир Олохтонов]

Coder 1448, а можете про контекст задачи немного рассказать? Это что-то учебное?

На первый взгляд идея запустить 2 event loop в разных процессах не выглядит безумной, но они не должны никак пересекаться по корутинам — всё-таки это разные рантаймы.

Тем не менее Сергей прав, так делать не принято, и тому есть причина — asyncio не дружит с форками, последняя попытка это сделать у меня упёрлась в дедлок.

Скорее всего ваша программа заработает если вы разделите сервер и клиент на 2 отдельных скрипта.

[Coder 1448]

Владимир Олохтонов, У меня тоже упирается в дедлок, при KeyboardInterrupt вижу что-то типо waiter.acquire()Задача реальная. Коротко - парсинг бирж с тяжелыми вычислениями. Каждую биржу решил вынести в отдельный процесс, чтоб процессор задействовать на 100%. А внутри каждого уже всё делать асинхронно. Появилась задача сделать прокси менеджер. Попытался расшарить синглтон между процессами, идея оказалась гиблой (кто сомневался). Решил сделать общение через сокеты.

Спасибо за совет, разделю на скрипты.

[Coder 1448]

Владимир Олохтонов, Решением стало каждый раз, когда посылаю сообщение, открывать новое соединение и в конце посылать EOF (так как я не часто посылаю сообщения). Был также вариант использовать readuntil с каким-нибудь сепаратором, но я от него отказался.