Как в python использовать функции async внутри multiprocessing.Pool?

Question

[Phoenix2000h]

with multiprocessing.Pool() as pool:
        results = pool.map(async_main, urls_chunks)

Когда пытаюсь закинуть в pool.map корутину, то получаю ошибку:

multiprocessing.pool.MaybeEncodingError: Error sending result: '[<coroutine object async_main at 0x7f3bac2e6bd0>, <coroutine object async_main at 0x7f3baacb5f10>, <coroutine object async_main at 0x7f3baacb6020>, <coroutine object async_main at 0x7f3baacb6130>]'. Reason: 'TypeError("cannot pickle 'coroutine' object")'

Т.е. использовать async функции c методами multiprocessing нельзя.
Есть ли способы обойти это ограничение или необходимо ждать новых версий python, или использовать другие ЯП (в том числе через возможности ОС, например bash, с запуском python файла в цикле)?

Comments

[Everything_is_bad]

зачем? ну так запускай event loop или ты вообще не понимаешь как работает asyncio?
а тег многопоточность к multiprocessing, говорит что ты и это не различаешь

[Phoenix2000h]

Everything_is_bad, Я понимаю, что multiprocessing связан с многоядерностью. Просто, модуль async работает в одном потоке, а хотелось бы использовать корутину одну и туже, сразу в нескольких потоках. Т.е. на каждом ядре через multiprocessing, например, шли асинхронные запросы к сайтам, а не только на 1 ядре, как есть в async.

[Everything_is_bad]

Phoenix2000h, еще раз, тогда тебе надо несколько event loop

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

Phoenix2000h, запросы к сайтам - это i/o bound задача, её бессмысленно распараллеливать по ядрам. Какую проблемы вы решаете?

[Phoenix2000h]

Сергей Горностаев, Есть рабочий код, в котором асинхронно делаются запросы к списку сайтов (из файла в нём указаны каждый на новой строке без пробелов):

import asyncio
import time
import cached
import httpx
from cachetools import cached, TTLCache

FILE_NAME = "/home/user/IdeaProjects/python_async_pytest/dir_async/urls.txt"


@cached(cache=TTLCache(maxsize=100, ttl=60))
async def fetch_url(url, queue_errors) -> (int, str):
    async with httpx.AsyncClient(timeout=10) as client:
        try:
            response = await client.get(url)
        except Exception as error:
            await queue_errors.put(url)
            return error
        else:
            return response.status_code, response.text


async def async_main(file_name):
    queue_errors = asyncio.Queue()
    start = time.time()

    with open(file_name, "r") as f:
        urls_before = f.read().splitlines()
    urls = ["https://www." + url for url in urls_before]

    tasks = [fetch_url(url, queue_errors) for url in urls]
    results = await asyncio.gather(*tasks)
    total_responses = len(results)
    end_time = round(time.time() - start, 2)

    urls_with_request_error = []
    for i in range(queue_errors.qsize()):
        urls_with_request_error.append(await queue_errors.get())
    print()

    print(end_time, "seconds")
    print("Total responses:", total_responses)
    print("Total successful responses:", total_responses - queue_errors.qsize())
    print("Total failed responses:", queue_errors.qsize())
    return results, urls_with_request_error


if __name__ == "__main__":
    asyncio.run(async_main(FILE_NAME))

Т.к. asynс разделяет 1 поток между работающими программами (корутинами), то пришла мысль сделать работу данной программы на каждом ядре и обойти ограничение в один поток, однако, столкнулся с ошибкой серилизации (писал выше в вопросе).
Возможно, я понимаю не всё (и/или не так) в параллельном программировании в python, поэтому буду удовлетворен ответом на вопрос ниже - можно ли ускорить эту программу выше путем реального распараллеливания (а не только конкурентного использования 1 потока) именно в python.

Answer 1

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

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

Comments

[Phoenix2000h]

Т.е. невозможно использовать на нескольких ядрах одну и туже async функцию, тем самым распаралеливая вычисления?
Покидание границы процессов - это, в частности, может быть запрос к внешнему, относительно ОС, ресурсу (сайту)?

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

Phoenix2000h, одну асинхронную функцию на разных ядрах запустить можно, каждый запуск породит сопрограмму выполняющуюся на конкретном ядре. Граница процесса - это выполнение в рамках одного интерпретатора. multiprocessing.Pool() создаёт несколько разных процессов-интерпретаторов, у каждого из них свой цикл событий и свои сопрограммы. Передавать сопрограмму от одного цикла в другой между процессами не просто технически невозможно, но и абсолютно бессмысленно.

Answer 2

[Vindicar]

Я очень надеюсь, что у тебя асинхронщина занимается получением данных, а multiprocessing - обработкой. В этом случае тебе не корутины по процессам надо раскидывать, а создавать внутри одной корутины синхронные задачи по обработке. А асинхронке оставить вопросы получение исходных данных/отправки результатов. Так куда практичнее. run_in_executor() в помощь.