Параллельные вызовы AI API зависают без 429: PoolTimeout HTTPX и закрытие потоков

Один запрос к AI API проходит, а несколько одновременных вызовов завершаются httpx.PoolTimeout. Причина не обязательно в замедлении провайдера: исключение означает, что клиент не получил соединение из пула за отведённое время. Это другой этап, чем HTTP 429 от сервера или ожидание данных по уже установленному соединению.

Материал относится к асинхронному клиенту HTTPX для Python. Проверка выполнена на Python 3.12.10, HTTPX 0.28.1 и локальном имитаторе HTTP/1.1. Реальные шлюзы, скорость моделей и платные API не тестировались. Использует ли конкретный SDK HTTPX и меняет ли настройки пула или повторов, нужно выяснять по его зависимостям и конфигурации.

1. Сохраните тип исключения и определите этап ожидания

Записи «тайм-аут запроса» недостаточно. HTTPX различает четыре вида тайм-аутов:[1]

Исключение или наблюдение	Где происходит ожидание	Что проверить первым
PoolTimeout	Получение соединения из клиентского пула	Активные потоки, незакрытые ответы, другие задачи и лимит пула
ConnectTimeout	Установление соединения с целевым хостом	Сеть, прокси, адрес и этап подключения
ReadTimeout	Получение следующего блока ответа	Пауза у провайдера, разрыв сети и тайм-аут чтения
WriteTimeout	Отправка следующего блока запроса	Загрузка, обратное давление и сеть
HTTP 429	HTTP-ответ уже получен	Какой сервис вернул 429, его квоты и правила ограничения

read=30 не ограничивает всю задачу тридцатью секундами: это ожидание блока данных. Длительный поток, регулярно передающий данные, может продолжаться гораздо дольше. Ожидание в очереди приложения тоже не входит в тайм-аут HTTPX pool; измеряйте эти уровни отдельно.

В отдельной прямой попытке передачи HTTPX тайм-аут пула возникает до получения соединения. Но одно нажатие кнопки может запускать несколько попыток в SDK, прокси или приложении. Последний PoolTimeout не доказывает, что вся задача никогда не отправлялась или не могла быть оплачена. Временное отсутствие записи в кабинете тоже недостаточно. Сохраняйте ID логической задачи и каждой попытки; при необходимости используйте шаблон обезличенной заявки и минимального воспроизведения.

2. Три параметра соединений — не три квоты параллелизма

Основные параметры HTTPX Limits:[2]

• max_connections: максимальное число соединений в пуле.
• max_keepalive_connections: число сохраняемых неактивных keep-alive-соединений, а не лимит активных запросов.
• keepalive_expiry: срок хранения неактивного соединения, а не тайм-аут генерации.

Повышение max_keepalive_connections с 5 до 50 не увеличивает автоматически параллелизм приложения с 5 до 50. Неограниченный max_connections также не расширяет квоту провайдера, зато способен увеличить локальное потребление ресурсов и нагрузку запросами.

В нашем эксперименте HTTP/1.1 один недочитанный поток занимает одно соединение. HTTP/2 позволяет передавать несколько потоков по одному соединению, поэтому равенство числа соединений и параллельных запросов нельзя переносить на него. Учитывайте реально согласованный протокол и ограничения потоков. Даже при http2=True запрос может идти по HTTP/1.1, если HTTP/2 не используется обеими сторонами. Проверяйте http_version ответа.[4]

Уточните область действия пула. Задачи одного клиента конкурируют за его ресурсы, а у каждого процесса и экземпляра клиента может быть собственный пул. Локальный лимит не является общей квотой всего развёртывания. Переиспользуйте клиент в пределах понятного жизненного цикла и закрывайте при завершении приложения. Не создавайте новый клиент для каждого запроса в частом цикле, чтобы «обойти» ожидание.[3]

3. Ответственность за закрытие потока должна быть явной

Долгие, но нормальные потоки могут заполнить пул так же, как забытые незакрытые ответы. Проверьте три пути:

1. Полное чтение: действительно ли ответ дочитан и проверено завершение прикладного протокола.
2. Ранний возврат или ошибка разбора: выйдет ли выполнение из контекста ответа.
3. Отмена пользователем или общий тайм-аут: выполняется ли очистка, а не только прекращение отображения.

HTTPX рекомендует async with client.stream(...): при выходе ответ закрывается. В ручном режиме client.send(..., stream=True) разработчик обязан обеспечить итоговый вызов await response.aclose().[3] Нельзя вернуть ещё не прочитанный поток из уже закрывшегося контекста, рассчитывая прочитать его позже в другом слое.

Закрытие недочитанного ответа HTTP/1.1 не гарантирует повторного использования того же TCP-соединения. Но оно должно освободить ресурс пула, занятый этим ответом, чтобы последующий запрос мог получить или установить соединение. Это также не гарантирует остановку удалённой модели: границы отмены и оплаты разобраны в материале о прекращении генерации и сверке списаний.

4. Воспроизведите ожидание пула локально, без настоящего API

Сохраните код как pool_demo.py и выполните python pool_demo.py в Python 3.11+ с установленным HTTPX. Скрипт слушает только временный порт 127.0.0.1 и не требует API Key.

Пул намеренно ограничен одним соединением, первый ответ останавливается на половине тела, а второй запрос ждёт всего 0,1 секунды. После выхода из контекста первого ответа новый запрос должен пройти. Ожидания сервера и клиента ограничены; в конце слушающий сокет и соединения закрываются.

import asyncio
import httpx


async def demo():
    seen = []
    release = asyncio.Event()
    handlers = set()

    async def handle(reader, writer):
        task = asyncio.current_task()
        handlers.add(task)
        try:
            header = await asyncio.wait_for(reader.readuntil(b"\r\n\r\n"), 1)
            path = header.split(b" ", 2)[1].decode("ascii")
            seen.append(path)
            writer.write(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n"
                         b"Connection: close\r\n\r\n")
            if path == "/hold":
                writer.write(b"a")
                await writer.drain()
                await asyncio.wait_for(release.wait(), 2)
                writer.write(b"b")
            else:
                writer.write(b"ok")
            await writer.drain()
        except (ConnectionError, TimeoutError, asyncio.IncompleteReadError):
            pass
        finally:
            writer.close()
            try:
                await writer.wait_closed()
            except ConnectionError:
                pass
            handlers.discard(task)

    server = await asyncio.start_server(handle, "127.0.0.1", 0, limit=4096)
    port = server.sockets[0].getsockname()[1]
    url = f"http://127.0.0.1:{port}"
    limits = httpx.Limits(max_connections=1, max_keepalive_connections=1)
    timeout = httpx.Timeout(1.0, pool=0.1)
    try:
        async with server, httpx.AsyncClient(
            limits=limits, timeout=timeout, http2=False,
            trust_env=False, follow_redirects=False
        ) as client:
            async with asyncio.timeout(5):
                async with client.stream("GET", url + "/hold") as response:
                    response.raise_for_status()
                    try:
                        await client.get(url + "/blocked")
                    except httpx.PoolTimeout:
                        print("PoolTimeout: waiting for the occupied slot")
                    else:
                        raise AssertionError("Expected PoolTimeout")
                result = await client.get(url + "/ok")
                assert result.status_code == 200 and result.text == "ok"
                assert seen == ["/hold", "/ok"], seen
                print("After stream close: 200; /blocked never reached server")
    finally:
        release.set()
        if handlers:
            await asyncio.gather(*list(handlers))


if __name__ == "__main__":
    asyncio.run(demo())

Ожидаются строки PoolTimeout: waiting for the occupied slot и затем After stream close: 200; /blocked never reached server. Дополнительная проверка подтверждает, что сервер получил только /hold и /ok.

Эксперимент не моделирует разбор ответа или полный протокол SSE и не доказывает утечку в конкретной производственной среде. Первый поток открыт намеренно; нормальные долгие потоки также заполняют маленький пул. trust_env=False исключает влияние прокси из окружения на локальный опыт. Это не рекомендация обходить необходимые производственные прокси или настройки сертификатов. Проверка TLS не отключается; сам опыт использует незашифрованный адрес обратной связи.

5. Объедините очередь, полный срок жизни потока и общий бюджет

Обычно исправление начинают с гарантированного закрытия ответов, затем ограничивают входящую нагрузку и только после измерений меняют размер пула и время ожидания. Увеличение pool лишь удлиняет очередь и не исправляет ответ, который никогда не закрывается.

Ниже функция чтения небольшого ответа; ей нужны импорты asyncio и httpx из предыдущего примера. Вызывающий код передаёт переиспользуемый в том же цикле событий клиент, общий asyncio.Semaphore и заранее проверенный фиксированный адрес сервиса. Функция выполняет GET и считает необработанные байты ответа; это не вызов модели и не SSE-парсер.

async def read_small(client, gate, url):
    async with asyncio.timeout(10):
        async with gate:
            async with client.stream("GET", url) as response:
                response.raise_for_status()
                size = 0
                async for chunk in response.aiter_raw():
                    size += len(chunk)
                    if size > 65536:
                        raise ValueError("response-size-limit")
                return size

Внешний asyncio.timeout(10) включает в бюджет функции и ожидание семафора. Он действует через кооперативную отмену, а не как точный принудительный предел в десять секунд: блокировка цикла событий и медленная очистка могут задержать завершение. Семафор должен охватывать всё чтение, а не освобождаться сразу после заголовков. Контекст тайм-аута Python преобразует вызванную им отмену в TimeoutError. Обрабатывайте его снаружи контекста и сохраняйте распространение отмены, не поглощая CancelledError.[5]

При превышении 65 536 байт функция выбрасывает исключение и закрывает ответ. Это предел счётчика необработанного тела, а не жёсткий лимит памяти процесса, проверка содержимого или размера после распаковки. Тело не накапливается, однако HTTPX уже мог передать большой блок данных. Обычная ошибка, тайм-аут и отмена вызывающей стороной должны приводить к очистке контекстов. Значения иллюстративные: подберите их под размер ответа, сроки и протокол.

Семафор ограничивает число задач внутри участка выполнения, но не число уже созданных ожидающих задач. Для пакетной обработки нужна ограниченная очередь или фиксированное число работников; при переполнении явно откладывайте или отклоняйте работу. Сотни тысяч заранее созданных корутин могут расходовать память даже за семафором. Размер пула HTTPX, параллелизм приложения и квоты RPM/TPM относятся к разным уровням и не заменяют друг друга.

6. При проверке исправления измеряйте этапы

Полезно записывать время постановки в очередь, получения разрешения приложения, начала вызова HTTPX, получения заголовков, окончания чтения или отмены; также класс исключения, идентификатор клиента и попытки. Длительности измеряйте монотонными часами, а для сопоставления с сервером сохраняйте время с часовым поясом. Не выводите ключи, полное тело запроса и URL с чувствительными параметрами.

Интервал от вызова HTTPX до заголовков включает ожидание пула, подключение, отправку и ожидание сервера. Называть весь интервал задержкой первого токена модели неправильно. Если данных транспортного уровня нет, укажите, что дальнейшее разделение невозможно.

Локальная регрессия должна включать заполненный долгим потоком пул, полное чтение, отмену в середине, тайм-аут чтения, HTTP-ошибку, превышение размера ответа и истечение общего бюджета в очереди приложения. После каждого сбоя отправляйте маленький фиксированный запрос тем же клиентом, проверяя восстановление разрешений и ресурсов соединений. Код 200 здесь подтверждает только ответ имитатора, а не завершение задачи реальной модели.

Оба примера прошли 12 локальных сценариев: перечисленные пути, пустой ответ, граничный размер, отказ от перенаправления и преждевременный обрыв тела. После ошибок проверено дальнейшее использование того же клиента и семафора. Это функциональная проверка на локальной сети обратной связи, а не измерение производительности или тест провайдера.

Затем наблюдайте допустимое окно небольшой рабочей нагрузки: новые PoolTimeout не появляются, очередь сокращается, завершения потоков согласуются с очисткой, скрытые повторы не усиливают нагрузку. Если проблема остаётся, проверьте другие задачи того же пула и жизненный цикл клиента. Один успешный запрос не даёт оснований объявлять провайдера стабильным или нестабильным.

7. Официальные источники и дата проверки

Проверено 2026-11-05. Примеры локально запускались на Python 3.12.10 / HTTPX 0.28.1; это версии проверочного окружения, а не заявление об актуальности последних выпусков.

1. HTTPX Timeouts: четыре вида тайм-аутов, ожидание блока данных и получение соединения; также сверена документация официального репозитория для 0.28.1.
2. HTTPX Resource Limits: общее число соединений, неактивные keep-alive-соединения и срок хранения.
3. HTTPX Async Support: переиспользование клиента, потоковый контекст и ручное закрытие; сверена документация репозитория для 0.28.1.
4. HTTPX HTTP/2: мультиплексирование, согласование и проверка фактического протокола.
5. Python asyncio Tasks: Timeouts и Task Cancellation: общий бюджет, отмена и очистка.