Что такое end-to-end backpressure в LLM пайплайне и как его реализовать?

Краткий тезис

End-to-end backpressure — это механизм согласования скорости обработки запросов на всех уровнях LLM-пайплайна, от API-шлюза до GPU. Без него перегрузка одного звена (например, очереди инференса) ведёт к потерям данных, нестабильности и падению качества сервиса. Реализация строится на bounded очередях, rate limiter'ах и сигналах HTTP 429 / gRPC RESOURCE_EXHAUSTED — каждый компонент передаёт давление вверх по потоку, чтобы система саморегулировалась.

1. Термин: Backpressure (обратное давление)

Backpressure — это принцип, при котором потребитель (consumer) сообщает производителю (producer) о своей текущей пропускной способности. Если потребитель перегружен, он сигнализирует «замедлись» или «остановись». В контексте LLM-пайплайна (цепочка: клиент → API gateway → очередь → vLLM → GPU) backpressure предотвращает переполнение буферов и обеспечивает предсказуемое поведение.

Зачем нужен backpressure

• GPU — дорогой и медленный ресурс. Если на него летит слишком много запросов — падает скорость генерации (batch latency растёт), возникает out-of-memory (OOM).
• Очереди без ограничений могут расти бесконечно → время ожидания становится неприемлемым.
• Без backpressure клиенты получают таймауты или непредсказуемые ошибки.

Термин «Bounded queue» (queue|ограниченная очередь) — очередь фиксированной ёмкости. При заполнении новые элементы не принимаются, вызывая отказ (например, Service Unavailable|HTTP 503). Это первый уровень backpressure.

2. Уровни LLM-пайплайна и точки pressure

End-to-end backpressure означает, что каждый уровень передаёт сигнал вышестоящему, создавая цепочку: GPU → vLLM → очередь → gateway → клиент.

3. Как работает backpressure на практике

3.1 API gateway → очередь

Gateway (например, Nginx с limit_req или Envoy с rate_limit) проверяет, не превышен ли лимит запросов в секунду. Если лимит превышен — отправляет HTTP 429 с заголовком Retry-After. Внутренне gateway может использовать bounded очередь к бэкенду (через upstream с max_conns).

3.2 Очередь → vLLM

Очередь (например, Kafka) имеет consumer group с настроенной max.poll.records и session.timeout.ms. Если consumer (vLLM) не успевает обрабатывать сообщения, Kafka смещает оффсет и блокирует чтение для данной партиции. Альтернатива: Redis-очередь с BLPOP и лимитом размера списка — при достижении лимита LPUSH возвращает ошибку.

3.3 vLLM → GPU

vLLM использует BoundedSemaphore на количество параллельно обрабатываемых последовательностей (--max-num-seqs). Когда лимит исчерпан, новые запросы попадают во внутреннюю очередь ожидания. Если и она переполняется, vLLM возвращает 503 Service Unavailable (в режиме --disable-frontend-multiprocessing).

3.4 Полный цикл обратного давления

1. GPU загружен — scheduler не принимает новые батчи.
2. vLLM внутренняя очередь заполняется → vLLM отвечает 503 на запрос от consumer (из очереди).
3. Consumer (например, Python worker) при получении 503 прекращает чтение из очереди на backoff секунд.
4. Очередь перестаёт опустошаться → её размер растёт.
5. Если размер достигает порога (например, 10k сообщений), gateway начинает отклонять новые запросы с 429.

Таким образом, давление распространяется от самого медленного звена (GPU) до клиента.

4. Реализация на конкретных инструментах

4.1 Kafka с квотами на consumer group

# Конфигурация consumer с backpressure
consumer = KafkaConsumer(
    'llm_requests',
    bootstrap_servers=['localhost:9092'],
    group_id='vllm-consumer',
    enable_auto_commit=False,
    max_poll_records=10,          # читать не больше 10 записей за poll
    max_poll_interval_ms=30000,   # если consumer занят дольше 30сек — партиция перераспределяется
    session_timeout_ms=45000
)

while True:
    records = consumer.poll(timeout_ms=1000)
    for topic_partition, messages in records.items():
        for msg in messages:
            # отправляем запрос в vLLM
            response = call_vllm(msg.value)  # может вернуть 503
            if response.status_code == 503:
                # backpressure: прекращаем poll на время
                time.sleep(5)
                consumer.pause(topic_partition)
                # ... дожидаемся восстановления и resume
            else:
                consumer.commit()

4.2 Redis-based rate limiter с Token Bucket

import redis
from ratelimit import limits, RateLimitException

r = redis.Redis()

# Проверка rate limit перед отправкой в vLLM
def check_rate_limit(user_id):
    current = r.get(f"rate:{user_id}")
    if current and int(current) > 100:  # 100 запросов в минуту
        raise RateLimitException("Too many requests")
    r.incr(f"rate:{user_id}", 1)
    r.expire(f"rate:{user_id}", 60)

Можно использовать bounded queue с Redis List: LLEN перед LPUSH — если длина > MAX, отказ.

4.3 vLLM настройка внутреннего scheduler

# Запуск vLLM с ограничениями
vllm serve meta-llama/Llama-2-7b-hf \
    --max-num-seqs 256 \
    --max-model-len 4096 \
    --disable-frontend-multiprocessing \
    --api-key your_key

Параметр --max-num-seqs ограничивает количество параллельных запросов. При превышении vLLM возвращает 503 с телом {"error":"Server is overloaded"}.

4.4 Global rate limiter на API gateway (Kong)

# Kong rate-limiting plugin
plugins:
  - name: rate-limiting
    config:
      second: 10
      hour: 10000
      policy: local  # или redis для распределённого
      fault_tolerant: true
      hide_client_headers: false

5. Протоколы сигналов backpressure

Экспоненциальный backoff — стратегия повторной отправки с увеличивающейся задержкой (1с, 2с, 4с, ...). Пример на Python:

import time
import requests
from requests.adapters import HTTPAdapter, Retry

session = requests.Session()
retries = Retry(total=5, backoff_factor=1, status_forcelist=[429, 503])
session.mount('http://', HTTPAdapter(max_retries=retries))

6. Мониторинг backpressure

Чтобы понимать, работает ли механизм, нужно отслеживать метрики:

• Queue length (размер очереди запросов) — Prometheus gauge.
• Request latency (p50, p99) — если растёт, значит backpressure не справляется.
• Rate of 429/503 responses — процент отказов.
• GPU utilization — если < 90%, возможно, backpressure слишком агрессивен.

Пример метрик в коде:

from prometheus_client import Gauge, Counter

queue_length = Gauge('request_queue_length', 'Current queue size')
rate_limit_rejected = Counter('rate_limit_rejected_total', 'Requests rejected by rate limiter')

7. Распространённые ошибки при реализации backpressure

1. Бесконечная очередь (unbounded queue) — из-за «на всякий случай». Приводит к OOM на брокере и неограниченному времени ответа.
2. Отсутствие backpressure для пакетных запросов — например, при streaming-генерации (Server-Sent Events) нельзя просто сбросить соединение; нужно дождаться окончания текущего ответа перед отказом.
3. Too granular backpressure — если каждый уровень слишком жёстко ограничивает, полезная пропускная способность падает. Важно настраивать лимиты на основе тестов.
4. Игнорирование backpressure на клиенте — без retry с backoff клиент будет постоянно слать запросы, перегружая gateway.

8. Сравнение подходов к реализации

9. Пет-проект для закрепления

Задача Реализовать симуляцию LLM-пайплайна с end-to-end backpressure на Python.

Инструменты asyncio, aiohttp, Redis (локально через Docker), prometheus_client.

Шаги:

1. Создать простой симулятор vLLM: async функция, которая принимает запрос и «обрабатывает» его с задержкой 0.5–2.0 с. Ограничение параллельных запросов asyncio.Semaphore(5).
2. Реализовать Worker, который читает из Redis-очереди (список) и отправляет запросы в симулятор. Если симулятор отвечает 503 (перегрузка), Worker засыпает на backoff.
3. Создать API Gateway на aiohttp с rate limiter (Token Bucket), который перед добавлением запроса в очередь проверяет длину очереди (LLEN). Если > 100, отвечает 429.
4. Написать скрипт-нагрузчик, который отправляет запросы с разной интенсивностью (1–50 req/s).
5. Вывести метрики: длина очереди, число 429, latency.

Ожидаемый результат Нагрузка на GPU-симулятор не превышает заданное число параллельных запросов; при всплесках клиенты получают 429, а не таймауты. Графики показывают стабильную латентность.

10. Связь с другими вопросами

11. Навигация

• Предыдущий: 428
• Следующий: 430
• Индекс: 00. Индекс разборов

Навигация

• Предыдущий: 428
• Следующий: 430
• Индекс: 00. Индекс разборов