Почему при подключениие через asyncio.open_connection в цикле плодятся потоки?

Question

[idin]

Друзья есть такой простой asyncio tcp клиент

import asyncio

async def tcp_echo_client(message):
    while True:
        try:
            reader, writer = await asyncio.open_connection('localhost', 9990)
        except OSError as e:
            print(f'OsError: {e}')
            await asyncio.sleep(5)
            continue
        while True:
            try:

                writer.write(message.encode())
                print(f'Send: {message!r}')
                data = await reader.read(4)
                print(f'Received: {data.decode()!r}')
                await asyncio.sleep(5)
            except  Exception as e:
                print(f'Error 2 while {e}')
                await asyncio.sleep(5)
                break
asyncio.run(tcp_echo_client('ping'))

Почему в цикле при пеерехвате ошибки OSError во время попытки подключении, когда например сервер не доступен (запускааем клеинт без серерва).
Я в htop вижу какието подпроцессы(ну или потоки) которые плодятся/ с каждым новым циклом.

Comments

[AWEme]

while True:
    try:
        reader, writer = await asyncio.open_connection('localhost', 9990)

Действительно, почему?

[idin]

AWEme я понимаю что он в цикле их создает , но как тогда мне их закрыть веди соиденения же не получилось ?

[idin]

reader writer не созданы

Answer 1

[Taus]

asyncio.open_connection порождает вызов socket.getaddrinfo в loop._default_executor, который в 3.7 по-умолчанию является ThreadPoolExecutor. Максимальное число воркеров в пуле по-умолчанию определяется max_workers = (os.cpu_count() or 1) * 5 (ссылка).
В 3.8 это изменят на min(32, (os.cpu_count() or 1) + 4), коммит.
Можете изменить число создаваемых потоков, передав свой экземпляр ThreadPoolExecutor:

import concurrent.futures
...
loop = asyncio.get_event_loop()
loop.set_default_executor(concurrent.futures.ThreadPoolExecutor(max_workers=4))
...

Comments

[idin]

а можно их как то закрывать ?

[Taus]

Зачем их закрывать? Их создаться максимальное количество и больше не будет.

[idin]

Taus, Может я не совсем понимаю как писать в этом ****ом асинхронном стиле)). Но чего то я вобще логики не понимаю.
Как мне контролировать эти потоки ThreadPool и зачем они вобще создаются. Тоесть например у меня в приложение помимо сокет-конекшена к серверу еще есть подключение к бд через

pool= await asyncpg.create_pool()

у которого минимум должно быть 10 этих пул-потоков. И та и та штука подключается к одному loop.
А я допустим поставил ограничение небольше 4-х, и получается что если клиент не сможет например какоето время подключится к серверу в тячении 10 попыток а потом подключился
Но затем идет подключение к бд но потоки уже забиты этими конекшенами к серверу и при этом их всего 4ре а надо как минимум 10 только для бд
И еще кстати когда происходит

await self.pool.close()

потоки почему то не умирают а так же висят в трее в htop.
Или это все подразумевает типо что все плодится автоматически и я недолжен забивать голову этим. Но это как то не правильно чтоли.
Ну или я неправильно делаю? Притом еще во всех книгах гвоорится что должен быть один loop на всю программу.

[idin]

Taus, я был бы очень благодарен если бы вы помогли понять как это работает и почему именно так. А то вроде как все класно, но ащущение что какаето костыльность у меня в коде

[Taus]

Это действительно сложные моменты. Вам стоит почитать по-подробнее про запуск долгого блокирующего кода в asyncio. В случае open_connection блокирующий вызов - это socket.getaddrinfo. Можете это проследить по коду в стандартной библиотеке. Если наивно вызвать эту функцию в корутине asyncio, то это приведёт к тому что asyncio-loop будет ждать выполнения этой функции перед передачей возможности выполнения другой корутине. Чтобы этого избежать используют Thread'ы. Для этого к asyncio-loop привязывается ThreadPoolExecutor для упрощения жизни разработчикам, позволяющий не следить самому за созданием и запуском потоков. Это позволяет запускать долгий блокирующий код loop.run_in_executor(...)в других потоках, которые не останавливают выполнение asyncio-loop в основном потоке на одной корутине. Пул потоков ThreadPoolExecutor не превышает размеры как я указал и при этом позволяет сэкономить на времени создания новых потоков.

asyncpg.create_pool создаёт пул соединений (проследите по вызовам), который конечно привязан к asyncio-loop, но не связан с пулом потоков ThreadPoolExecutor.

[idin]

Taus, Спасибо за ответ я почитал про loop.run_in_executor(...), вроде понял что это сделанно для того чтобы сделать блокирующие функции неблокирующими благодоря переноса задач в треды (даже попробывал свои функции позапихивать туда). И как я понял эти пулы как то сами там между собой делят задачи когда выполняются (тут я написал для себя код чтобы проверить это):

import asyncio
import asyncpg
# from config import ConfServer
import time
import concurrent.futures
from dataclasses import dataclass
import struct

@dataclass(order=True)
class ConfServer:
    db_user: str = "admin"
    db_password: str = 'vrqegtq4r23ewqs'
    db_name: str = 'chat'
    db_host: str = 'localhost'

config = ConfServer()


async def readmsg(reader):
    d = await asyncio.wait_for(reader.readexactly(4), timeout=30)
    d = struct.unpack('>I', d)[0]
    d = await asyncio.wait_for(reader.readexactly(d), timeout=30)
    d = d.decode("utf-8")
    if d == "pong":
        return "pong"
    return d

def time_sleep():
    if False:
        time.sleep(5)


async def close_ppool(pool):
    try:
        await pool.close()
        print('create_pool close')
    except:
        print('except pool.close')
        time_sleep()

async def run():
    while True:
        try:
            print('start create_pool')
            pool = await asyncpg.create_pool(user=config.db_user,
                                          password=config.db_password,
                                          database=config.db_name,
                                          host=config.db_host)
        except:
            print('except create_pool')
            time_sleep()
            continue



        try:
            print('start open_connection')
            reader, writer = await asyncio.open_connection('localhost', 9990)
        except OSError as e:
            print("OSError : tcp_client_start")
            time_sleep()
            await close_ppool(pool)
            continue

        writer.write(struct.pack('>I', 4) + 'ping'.encode("utf-8"))

        data = await readmsg(reader)
        print("DATA <<<:  " + str(data) + "\n")
        writer.close()
        await writer.wait_closed()

        await close_ppool(pool)
        time.sleep(5)




loop = asyncio.get_event_loop()
loop.set_default_executor(concurrent.futures.ThreadPoolExecutor(max_workers=0))
loop.run_until_complete(run())

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

В Целом вроде понял идею работы.

Но какое-то не допонимание есть, в плане того как правильно написать этот асинхронны код.

Можно ли как то с вами списаться по поводу этого (получить какой нибудь ваш контакт), и узнать правильно ли я пишу код ну или небольшие коментарии к нему, в ту ли сторону я мыслю)). Или я не праильно понял логику работы asyncio. Я обещаю постораться спрашивать исключительно если прям сам несмог найти ответ на свой вопрос.
Ну или можете написать мне в телегу @izy_di.

[volkodav_kel]

Taus, к вопросу "А зачем меньше", когда ты создашь под 400 процессов на 32х ядерной машине то поймешь мою боль))

Благодарочка, помог разобраться с этим ...