Тяжёлые расчёты в Python API: отдельный пул процессов, предел очереди и замер ожидания

Представьте обычный сервис на Python: десяток лёгких эндпоинтов, один из которых принимает параметры и считает сложную модель — оптимизацию маршрутов, свёртку большого массива, генерацию отчёта с тяжёлой статистикой. Пока расчёт никто не запускает, всё хорошо. Затем приходит один клиент с крупным входом, и вдруг /health, который раньше отвечал за пару миллисекунд, начинает отвечать за секунды. Балансировщик решает, что сервис мёртв, и выводит его из ротации. Сервис при этом жив и усердно считает; он просто не успевает сказать об этом вслух.

Эта статья о том, как разделить такой сервис на части, чтобы тяжёлая работа не ломала лёгкую, и о менее очевидной стороне вопроса: что делать, когда тяжёлой работы приходит больше, чем сервер способен переварить. Вынести расчёт в отдельные процессы — только половина решения. Вторая половина — не позволить очереди перед этими процессами расти без границ.

Три роли, которые не стоит совмещать в одном запросе

В наивной версии один HTTP-запрос делает всё сразу: принимает параметры, считает и тут же возвращает ответ. Три разные обязанности склеены в один вызов, и у каждой свои требования ко времени.

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

Пока эти роли слиплись, поведение сервиса определяется худшей из них. Клиент с медленным соединением держит открытый запрос; расчёт держит процессор; а балансировщик со своим таймаутом обрывает соединение в тот момент, когда работа почти доделана, так что результат уходит в никуда. Разделение даёт каждой части собственные правила. Приём — быстрый и способный сказать «нет». Исполнение — ограниченное по числу одновременных задач. Выдача результата — обычное чтение из хранилища состояний.

На уровне протокола это превращается в привычную схему: POST /jobs отвечает кодом 202 и идентификатором, GET /jobs/{id} показывает состояние и результат, DELETE /jobs/{id} отменяет то, что ещё можно отменить. Не обязательно использовать именно эти пути, важно само разделение. Для совсем коротких расчётов можно оставить синхронный вариант с ожиданием — об этом мы поговорим ниже, потому что он тоже подчиняется описанным правилам.

Сто корутин — это всё ещё один процессор

Первая реакция на медленный эндпоинт в асинхронном фреймворке часто такая: функция уже объявлена как async def, значит, она неблокирующая. Это не так. Корутина уступает управление циклу событий только в точках await, где она ждёт что-то внешнее: ответ базы, сетевой пакет, таймер. Если же внутри цикла считаются числа, то никакого await нет, и пока вычисление не закончится, цикл событий не обслуживает никого. Ни новые соединения, ни /health, ни чужие корутины.

Отсюда следует простое различие. Ожидание ввода-вывода хорошо масштабируется корутинами: тысяча запросов, ждущих ответов от внешнего API, почти не потребляют процессор, и цикл переключается между ними. CPU-bound работа — это когда процессор занят самим вычислением. Добавление корутин здесь не создаёт ни одной дополнительной секунды процессорного времени: все они стоят в очереди к одному ядру, на котором работает цикл, и каждая, дорвавшись до управления, не отпускает его до конца.

Следующая мысль — отправить расчёт в поток. В документации concurrent.futures описаны ThreadPoolExecutor и ProcessPoolExecutor, которые исполняют вызовы в потоках и в отдельных процессах соответственно (Python: concurrent.futures). Для задач, где поток большую часть времени ждёт, потоки подходят. Для чистого Python-кода, который считает, в стандартной сборке CPython картина иная: потоки делят интерпретатор и один блокирующий механизм исполнения байт-кода, поэтому параллельного ускорения на нескольких ядрах вы не получите, а событийный цикл при этом будет конкурировать с этим потоком за интерпретатор. Это поведение конкретной реализации и конкретной сборки, и переносить его на любой Python runtime нельзя: есть сборки и расширения, где картина отличается. Нативные библиотеки вроде численных могут отпускать блокировку на время тяжёлых операций, и тогда потоки работают лучше. Не гадайте — на вашем коде и вашем интерпретаторе это решается коротким замером: запустите одну и ту же функцию в четырёх потоках и в четырёх процессах и сравните общее время.

Отдельные процессы дают то, чего не дают корутины и (в типичном случае) потоки: у каждого процесса свой интерпретатор и своё ядро, на котором он может крутиться независимо. Цена — передача данных через сериализацию. Аргументы функции и результат пересылаются между процессами, значит, они должны сериализоваться, и пересылка огромного массива может съесть часть выигрыша. Поэтому в рабочий процесс лучше передавать компактное описание задания (параметры, путь к файлу), а не гигабайтный объект.

Пул процессов с потолком на количество выполняемых задач

Чтобы вынести расчёт, удобно взять ProcessPoolExecutor. Метод submit возвращает Future, описывающий выполнение (документация concurrent.futures), а из асинхронного кода пул подключается через loop.run_in_executor. Количество процессов задаётся параметром max_workers: это и есть верхняя граница одновременно выполняемых расчётов. Пока эта цифра равна четырём, пятое задание физически не начнёт работу раньше, чем освободится один из процессов.

Но вот что важно: потолок на выполняемые задачи — это ещё не потолок на очередь. Если принимать каждое входящее задание и просто отправлять его в пул, то излишек не исчезнет, а начнёт копиться где-то внутри. Пул не обязан сам отказывать клиентам; решать, сколько ожидающих заданий допустимо, должен ваш код. Поэтому в примере ниже ожидающие задания хранятся не внутри пула, а в вашей собственной очереди, и пул получает работу только тогда, когда один из его процессов готов её взять.

Вот компактный пример на asyncio: четыре исполнителя, не больше восьми ожидающих, предельный возраст ожидания — пять секунд. Все числа условные, подбираются под вашу задачу.

import asyncio, time, uuid
from concurrent.futures import ProcessPoolExecutor

WORKERS = 4
MAX_WAITING = 8
MAX_QUEUE_AGE = 5.0

class Busy(Exception):
    pass

def heavy(payload):
    # выполняется в дочернем процессе: чистая функция от входа
    return sum(i * i for i in range(payload['n']))

class Job:
    def __init__(self, payload):
        self.id = uuid.uuid4().hex
        self.payload = payload
        self.state = 'queued'
        self.accepted_at = time.monotonic()
        self.started_at = None
        self.finished_at = None
        self.result = None
        self.error = None

class Compute:
    def __init__(self):
        self.jobs = {}
        self.waiting = 0

    async def start(self):
        self.pool = ProcessPoolExecutor(max_workers=WORKERS)
        self.queue = asyncio.Queue()
        self.runners = [asyncio.create_task(self._runner()) for _ in range(WORKERS)]

    def submit(self, payload):
        if self.waiting >= MAX_WAITING:
            raise Busy()
        job = Job(payload)
        self.jobs[job.id] = job
        self.waiting += 1
        self.queue.put_nowait(job)
        return job

    def cancel(self, job_id):
        job = self.jobs.get(job_id)
        if job and job.state == 'queued':
            job.state = 'cancelled'
            self.waiting -= 1
            return True
        return False

    async def _runner(self):
        loop = asyncio.get_running_loop()
        while True:
            job = await self.queue.get()
            if job.state != 'queued':
                continue
            self.waiting -= 1
            now = time.monotonic()
            if now - job.accepted_at > MAX_QUEUE_AGE:
                job.state = 'expired'
                job.finished_at = now
                continue
            job.state = 'running'
            job.started_at = now
            try:
                job.result = await loop.run_in_executor(self.pool, heavy, job.payload)
                job.state = 'done'
            except Exception as exc:
                job.state = 'failed'
                job.error = type(exc).__name__
            finally:
                job.finished_at = time.monotonic()

Что здесь происходит и почему так. Исполнителей-корутин ровно столько же, сколько процессов в пуле, и каждая в любой момент держит не более одного задания. Значит, в пул одновременно уходит не больше четырёх задач, и никакой скрытой очереди внутри него не образуется. Всё ожидание живёт в нашем asyncio.Queue, где мы его видим, считаем и можем отменять.

Проверка self.waiting >= MAX_WAITING и последующее увеличение счётчика выполняются без единого await между ними. Цикл событий однопоточный, и другая корутина не может вклиниться посередине, поэтому дополнительная блокировка не нужна. Это свойство конкретной схемы с одним циклом: если вы будете вызывать submit из нескольких потоков, потребуется синхронизация.

Функция heavy принимает простой словарь и возвращает число. Она определена на уровне модуля, потому что вызываемый объект пересылается в рабочий процесс, а не создаётся там заново. Лямбды и вложенные функции для этого, как правило, не подходят.

Теперь про одну ловушку, которую в документации выделяют отдельно: ожидание одной Future внутри задачи, уже занявшей нужного исполнителя, может привести к deadlock (Python: concurrent.futures). Для нашей схемы практический вывод такой: не заставляйте рабочую функцию внутри пула отправлять подзадания в тот же пул и ждать их. Если все четыре процесса заняты такими «родителями», дети никогда не получат места. Хочется разбить крупный расчёт на части — разбейте его снаружи, в обработчике очереди, а не изнутри рабочего процесса.

Когда ожидание заполнилось: честный отказ вместо вежливой очереди

Теперь главный вопрос: что делать с девятым ожидающим заданием? Варианты ровно три: принять и надеяться, заставить клиента ждать на соединении или сказать «занято».

Покажем на условном примере, почему первый вариант хуже, чем кажется. Допустим, одно задание занимает процесс примерно две секунды, а процессов четыре. Тогда система обрабатывает порядка двух заданий в секунду. Если приходят три в секунду, очередь растёт на одно задание каждую секунду. Через минуту в ней шестьдесят заданий, и новое, поставленное в хвост, дождётся своей очереди примерно через полминуты, а свои собственные две секунды вычислений получит только после этого. Клиент с таймаутом в десять секунд давно ушёл. Ваши процессы при этом честно считают результаты, которые уже никто не прочитает, а новые клиенты стоят за ними. Бесконечная очередь не защищает от перегрузки, она превращает её в постоянное состояние с большой задержкой.

С ограничением картина иная. При четырёх процессах и восьми ожидающих заданиях и двух секундах на задание наихудший случай для принятого задания — дождаться, пока освободятся два «круга» по четыре задачи: это около четырёх секунд ожидания и ещё две на собственный расчёт. Верхняя граница известна заранее, и её можно сообщить клиенту. А всё, что не влезло, получает немедленный отказ.

Для HTTP это естественно выражается кодом 503 и заголовком Retry-After с разумной паузой. Некоторые предпочитают 429, но он по смыслу про превышение лимита конкретным клиентом, а здесь перегружен сервер в целом, так что 503 честнее. Тело ответа должно быть коротким и машинночитаемым: сервис занят, попробуйте через столько-то секунд. Обработчик приёма в таком случае выглядит так:

@app.post('/jobs', status_code=202)
async def create_job(body: dict):
    try:
        job = compute.submit(body)
    except Busy:
        return JSONResponse({'error': 'busy'}, status_code=503, headers={'Retry-After': '3'})
    return {'id': job.id, 'state': job.state}

Здесь нужно отметить, что отказ — это не поломка, а рабочий режим. Клиент, получивший 503 за несколько миллисекунд, может немедленно решить: подождать, попробовать другой экземпляр, показать пользователю сообщение. Клиент, чей запрос висел шестьдесят секунд и закончился таймаутом, не знает, выполнится ли его задание вообще. Быстрый отказ — это информация, а затянувшееся молчание — нет.

Второй параметр очереди — время жизни задания. В примере это MAX_QUEUE_AGE: если задание ждало дольше пяти секунд, его уже не стоит запускать, потому что клиент либо ушёл, либо получит результат слишком поздно. Запуск такого задания тратит ресурс впустую. Значение выбирается из требований клиента, а не из возможностей сервера: если пользователь готов ждать до десяти секунд, а расчёт занимает две, предельный возраст не должен превышать примерно шести-семи секунд. Заодно это страхует вас от неточностей в размере очереди: даже если ограничение выбрано слишком щедрым, устаревшее задание отсеется при выходе из очереди.

Ожидающее задание, которое никому не нужно

Теперь самый тонкий момент. Клиент отправил задание, оно стоит в очереди, и тут клиент передумал: закрыл вкладку, нажал «Отмена», его собственный таймаут сработал. Что должно произойти с заданием?

Ответ зависит от того, на каком этапе оно находится, и в нашей схеме у него есть три принципиально разных состояния.

• queued — ждёт в нашей очереди. Работа ещё не передана в пул, поэтому отмена бесплатна: достаточно пометить задание отменённым и освободить место. Когда исполнитель вынет его из очереди, он увидит состояние cancelled и пропустит его. Обратите внимание, что мы уменьшаем счётчик waiting сразу при отмене, а не при извлечении из очереди. Иначе отменённые задания продолжали бы занимать места, и новые клиенты получали бы отказ из-за призраков.
• running — уже считается в рабочем процессе. Здесь простой отмены нет. Вызов, который уже выполняется в дочернем процессе, остановить из родителя аккуратно нельзя: можно лишь перестать ждать результат. Для пула процессов это означает, что процесс доведёт расчёт до конца, а результат будет выброшен. Поэтому честная модель такая: отмена работающего задания либо запрещена, либо означает «результат не нужен, но процесс освободится только после завершения». Если расчёты настолько длинные, что это неприемлемо, вам нужны проверки флага отмены внутри самой функции или отдельные процессы, которые можно завершить, но это уже другая конструкция, и её цена — потеря состояния такого процесса.
• done, failed, expired, cancelled — финальные состояния. Они ничего не потребляют, кроме памяти под запись о результате.

Эта разница объясняет, почему отдельный счётчик ожидающих и собственная очередь лучше, чем «отправим всё в пул и вызовем cancel() у Future». Для Future смысл отмены зависит от того, началось ли выполнение, и вы всё равно будете разбирать эти ветки; плюс вы не видите внутреннюю очередь пула и не можете считать места. В нашей схеме граница между «ещё можно отменить бесплатно» и «уже поздно» проходит ровно там, где вы сами её провели: в момент вызова run_in_executor.

Есть ещё один случай, в котором отмена происходит без участия клиента: срок жизни. Задание, ждавшее дольше MAX_QUEUE_AGE, переходит в expired при извлечении. Можно делать это и раньше, периодической проверкой хвоста очереди, но в простом случае отложенная проверка достаточна: пока все исполнители заняты, устаревшее задание никому не мешает, кроме счётчика мест. Если для вас это существенно, добавьте фоновую корутину, которая раз в секунду помечает просроченные задания и уменьшает waiting.

Результат на потом и обрыв связи посреди расчёта

В схеме с 202 и опросом клиенту не нужно держать соединение, поэтому обрыв связи перестаёт быть событием, угрожающим заданию. Клиент пропал на минуту, вернулся, спросил GET /jobs/{id} и получил done с результатом. Расчёт и доставка отделены друг от друга, и именно в этом ценность схемы.

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

1. Рост без границ. Если не удалять старые записи, словарь заданий сам превратится в утечку. Нужен срок хранения (скажем, несколько минут после завершения) и периодическая очистка. Результат, который клиент не забрал вовремя, удаляется; на запрос по такому идентификатору честный ответ — 404 или 410.
2. Несколько веб-процессов. Если приложение запущено в нескольких воркерах веб-сервера, у каждого будет своя очередь, свой пул и своя память. Клиент, отправивший задание в одного воркера, а опрашивающий другого, получит «не найдено». Кроме того, потолок нужно считать суммарно: четыре воркера по четыре процесса — это шестнадцать расчётных процессов, а не четыре. Подробнее про устройство воркеров смотрите в статье VPS для FastAPI: uvicorn и gunicorn workers. В нашей схеме проще всего держать один веб-процесс на вычислительный пул, либо вынести вычисления в отдельный сервис с единой очередью.
3. Перезапуск. Задания в ProcessPoolExecutor не сохраняются при падении или перезапуске сервера: очередь и состояние лежат в памяти. После рестарта идентификаторы, выданные ранее, перестанут существовать. Если для вас это неприемлемо, нужно внешнее хранилище состояний и внешняя очередь — это уже другой уровень сложности, и ничто из показанного здесь его не заменяет. Зато честная формулировка для клиента проста: «задание не найдено — отправьте заново», при условии, что расчёт безопасно повторять.

А если клиенту нужен синхронный ответ, потому что расчёт обычно занимает долю секунды? Тогда можно оставить один эндпоинт, внутри которого вы вызываете submit и ждёте задание с таймаутом. Правила остаются теми же: тот же потолок очереди, тот же 503 при переполнении. Но появляется вопрос про обрыв: если клиент отвалился, пока задание стоит в очереди, его стоит отменить. В большинстве асинхронных фреймворков обрыв соединения проявляется как отмена корутины обработчика, и в блоке finally можно вызвать compute.cancel(job.id). Если задание уже выполняется, вызов вернёт False, и расчёт доработает впустую. Здесь удобно иметь счётчик таких «осиротевших» запусков: если он растёт, значит, ваши таймауты на клиентской и серверной стороне не согласованы.

Состояния, которые клиент может увидеть, в итоге такие:

Измерять ожидание отдельно от вычисления

В примере у каждого задания три метки времени: accepted_at, started_at, finished_at. Это не декорация: они позволяют разделить две величины, которые при общем замере «запрос шёл столько-то секунд» неразличимы.

• Время ожидания = started_at − accepted_at. Показывает, насколько система перегружена. Растёт при росте притока, не зависит от того, насколько быстра сама функция.
• Время вычисления = finished_at − started_at. Показывает стоимость самой работы. Растёт при тяжёлых входах или при нехватке ядер, на которых процессы реально идут параллельно.

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

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

Другая полезная метрика — задержка лёгких эндпоинтов. Это и есть тест на то, что вынос расчёта удался: пока кто-то отправляет тяжёлые задания, /health должен отвечать так же быстро, как и в тишине. Запустите на тестовом стенде нагрузку на тяжёлый эндпоинт и параллельно меряйте /health. Если его задержка выросла заметно, значит, что-то по-прежнему блокирует цикл: например, десериализация огромного входа или сборка огромного результата в родительском процессе.

Сколько процессов поставить на VPS

Для CPU-bound работы число рабочих процессов упирается в число ядер. Если процессов больше, чем vCPU, они начинают делить ядра, время каждого расчёта растёт, а суммарная пропускная способность почти не меняется. Если меньше, часть ядер простаивает. Разумная отправная точка — число vCPU минус то, что нужно самому веб-процессу и системе. Но это именно точка отсюда начать, а не готовый ответ: окончательное число показывают замеры вашей функции на вашей машине.

В публичном предложении VPS MAATRIX на NVMe с процессорами AMD EPYC и root-доступом перечислены конфигурации от 1 vCPU и 2 ГБ памяти (Ba) до 16 vCPU и 32 ГБ (Djed); между ними Uab с 2 vCPU и 4 ГБ, Heka с 4 vCPU и 8 ГБ, Maat с 8 vCPU и 16 ГБ. Эти цифры говорят о том, сколько ресурсов выделено виртуальной машине, и не обещают определённого числа одновременных клиентов или расчётов в секунду: сколько заданий потянет ваш сервис, зависит от самого вычисления. Практический смысл такой: на конфигурации с одним vCPU выделять пул из четырёх процессов бессмысленно, лучше один процесс и маленькая очередь; на четырёх vCPU начинайте с трёх процессов и оставляйте ядро веб-части; а для восьми и более vCPU проверяйте, не упирается ли расчёт в память, потому что каждый процесс держит собственную копию данных.

Память — вторая половина уравнения. Каждый рабочий процесс — это отдельный интерпретатор с собственными структурами. Если функция при расчёте раздувается до гигабайта, четыре процесса потребуют четыре гигабайта, и тесная конфигурация уйдёт в своп раньше, чем закончатся ядра. Поэтому потолок на количество выполняемых задач — это в том числе и предохранитель по памяти. Прежде чем увеличивать WORKERS, посмотрите пиковое потребление одного процесса под реальным входом.

Доступ к машине с правами root тут нужен по делу: вы сможете смотреть загрузку по ядрам, пиковую память отдельных процессов и подобрать число рабочих процессов под реальное поведение функции, а не угадывать по спецификации. Если сомневаетесь в выборе, начните с меньшей конфигурации, снимите метрики ожидания и вычисления и перейдите на следующую, когда данные покажут упор именно в ядра, а не в память или в неудачный код.

Когда лучше сказать «занято»

Вернёмся к началу, к сервису, у которого простые запросы замедлялись из-за тяжёлых. После всех изменений его поведение выглядит иначе. Тяжёлый расчёт приходит, получает идентификатор за несколько миллисекунд и уходит в ограниченный пул процессов. Лёгкие эндпоинты отвечают как обычно, потому что цикл событий занят только приёмом и выдачей. Метки времени у каждого задания позволяют отличить перегрузку от дорогого входа.

А что происходит в час пик? Приходит больше заданий, чем способны пережить четыре процесса за разумное время. Очередь из восьми мест заполнена, и девятое задание получает короткий ответ: сервис занят, повторите через три секунды. Это раздражает, но это правда: за время ожидания сервер всё равно не выполнил бы работу в срок, а клиент либо ушёл бы, либо получил бы результат, которого уже не ждал. Принять задание — значит взять на себя обязательство. Сервер, который берёт работу, зная, что не успеет, обещает то, чего не может выполнить, и расплачивается за это чужими простаивающими процессорами и растущей очередью.

Честный отказ превращает перегрузку из скрытой задержки в явный сигнал. Клиент может подождать, балансировщик может направить запрос на другой экземпляр, мониторинг покажет всплеск отказов, и вы поймёте, пора ли расширять машину или пересматривать расчёт. Очередь, у которой есть дно, делает систему предсказуемой: вы заранее знаете, сколько заданий она держит и сколько секунд готова ждать, и говорите «занято» ровно тогда, когда это правда.