Bài 51: Batching — gom nhiều request để inference 1 lần

Sau bài này bạn sẽ:

• ✅ Hiểu vì sao batching tăng GPU throughput và giảm cost-per-request
• ✅ Phân biệt static batching, dynamic batching, continuous batching
• ✅ Tự implement DynamicBatcher với asyncio trong FastAPI
• ✅ Biết dùng vLLM, Triton, TorchServe cho managed batching
• ✅ Dùng OpenAI/Anthropic Batch API cho workload offline
• ✅ Tránh các lỗi phổ biến: OOM, SLO violation, lock contention

GPU có hàng nghìn CUDA core nhưng 1 request đơn lẻ chỉ chiếm một phần nhỏ trong số đó. Phần còn lại idle trong khi đợi. Khi gộp B request thành 1 batch:

• Kernel launch overhead chia sẻ: 1 lần launch kernel thay vì B lần.
• Weight load 1 lần: weights model load từ HBM (High Bandwidth Memory) vào L2/L1 cache một lần, dùng cho toàn batch.
• Parallelism theo chiều batch: phép nhân ma trận W × X với X là batch → GEMM (General Matrix Multiply) vectorized trên toàn GPU.

Kết quả thực tế (GPU A100, BERT-base, fp16):

Throughput tăng gần tuyến tính đến khi GPU bão hòa (memory bandwidth hoặc compute bound). Latency mỗi request tăng nhẹ do request phải chờ đủ batch trước khi xử lý — đây là trade-off cốt lõi của batching.

Static batching: client gửi sẵn một list input trong 1 request. Server nhận, chạy model 1 lần, trả về list kết quả.

# models.py
from pydantic import BaseModel

class BatchRequest(BaseModel):
    texts: list[str]

class EmbedRequest(BaseModel):
    text: str

# app.py — FastAPI 0.110+
from fastapi import FastAPI
from sentence_transformers import SentenceTransformer

app = FastAPI()
model = SentenceTransformer("BAAI/bge-small-en-v1.5")

@app.post("/embed_batch")
def embed_batch(req: BatchRequest):
    # req.texts: list[str], gom sẵn từ client
    embeddings = model.encode(req.texts, batch_size=32, normalize_embeddings=True)
    return {"embeddings": embeddings.tolist()}

Client gọi:

import httpx

texts = ["sentence A", "sentence B", "sentence C"]  # gom trước
resp = httpx.post("http://localhost:8000/embed_batch", json={"texts": texts})
embeddings = resp.json()["embeddings"]

Ưu và nhược điểm

Ưu: đơn giản, performance cao nếu client kiểm soát được batch.

Nhược: nhiều client độc lập, mỗi client gửi 1 request riêng → server nhận từng request đơn lẻ, không batch được. Đây là tình huống phổ biến trong production khi có N microservice hoặc N user đồng thời gọi API.

Dynamic batching: server tự gom các request đến trong một time window. Mỗi request submit item vào queue, một background worker thu thập và chạy batch khi đủ batch_size hoặc hết max_wait_ms.

import asyncio
from collections import deque
from sentence_transformers import SentenceTransformer
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()
model = SentenceTransformer("BAAI/bge-small-en-v1.5")


class EmbedRequest(BaseModel):
    text: str


class DynamicBatcher:
    def __init__(self, batch_size: int = 32, max_wait_ms: float = 20):
        self.batch_size = batch_size
        self.max_wait_ms = max_wait_ms
        # Dùng asyncio.Queue thay vì deque + Lock để tránh lock contention
        self.queue: asyncio.Queue = asyncio.Queue()

    async def start(self):
        asyncio.create_task(self._worker())

    async def submit(self, item: str) -> list[float]:
        future: asyncio.Future = asyncio.get_event_loop().create_future()
        await self.queue.put((item, future))
        return await future

    async def _worker(self):
        while True:
            # Chờ item đầu tiên
            item, fut = await self.queue.get()
            batch = [item]
            futures = [fut]

            # Thu thêm item trong max_wait_ms hoặc đến khi đủ batch_size
            deadline = asyncio.get_event_loop().time() + self.max_wait_ms / 1000
            while len(batch) < self.batch_size:
                remaining = deadline - asyncio.get_event_loop().time()
                if remaining <= 0:
                    break
                try:
                    item, fut = await asyncio.wait_for(
                        self.queue.get(), timeout=remaining
                    )
                    batch.append(item)
                    futures.append(fut)
                except asyncio.TimeoutError:
                    break

            # Xử lý batch
            try:
                results = await asyncio.to_thread(
                    model.encode, batch, normalize_embeddings=True
                )
                for fut, res in zip(futures, results):
                    fut.set_result(res.tolist())
            except Exception as exc:
                # Fail từng future riêng, không fail toàn batch silently
                for fut in futures:
                    if not fut.done():
                        fut.set_exception(exc)


batcher = DynamicBatcher(batch_size=32, max_wait_ms=20)


@app.on_event("startup")
async def startup():
    await batcher.start()


@app.post("/embed")
async def embed(req: EmbedRequest):
    result = await batcher.submit(req.text)
    return {"embedding": result}

Với implementation này, 50 client gọi /embed đồng thời trong cùng 20ms window sẽ được gom vào 1–2 batch thay vì 50 lần forward riêng lẻ.

Lưu ý về asyncio.Queue vs deque + Lock: asyncio.Queue là thread-safe trong single-thread event loop và tránh được lock contention. Dùng deque + asyncio.Lock trong batcher đơn giản thường gây vấn đề vì acquire lock trong hot path.

Hai tham số chính cần tune:

• batch_size: giới hạn trên số item per batch. Tăng → throughput tăng, latency tăng (compute lâu hơn), nguy cơ OOM.
• max_wait_ms: thời gian chờ tối đa để thu đủ batch. Tăng → batch fill rate tốt hơn khi traffic thưa, nhưng mọi request đều chờ ít nhất max_wait_ms.

Cách tune thực tế:

1. Xác định SLO: ví dụ P95 latency < 200ms.
2. Load test với traffic profile thực tế (peak QPS, off-peak QPS).
3. Đo P50/P95/P99 latency ở các giá trị max_wait_ms = 5, 10, 20, 50.
4. Chọn giá trị cao nhất mà P95 vẫn dưới ngưỡng SLO ở peak traffic.

Quy tắc thực tế:

• Embedding model, không realtime: max_wait_ms=20–50, batch_size=32–64.
• Chat inference realtime: max_wait_ms=5–10, batch_size=8–16.
• Traffic thấp (dev/staging): max_wait_ms lớn sẽ làm mọi request chờ — nên giảm xuống 5ms.

Static batching áp dụng cho LLM (autoregressive generation) có vấn đề cơ bản: mỗi sequence trong batch sinh ra số token khác nhau. Batch phải đợi sequence dài nhất hoàn tất → các sequence ngắn đã xong nhưng vẫn chiếm slot, GPU tính toán "padding token" vô ích.

Continuous batching (Orca, 2022 — Yu et al., arXiv:2207.04869) giải quyết bằng cách hoạt động ở mức token step thay vì request level:

• Mỗi iteration (1 token decode step), scheduler kiểm tra sequence nào đã done (gặp EOS).
• Swap sequence done ra, swap sequence mới từ queue vào ngay trong iteration đó.
• Batch luôn gần đầy, không còn slot idle chờ sequence dài.

Kết quả: throughput tăng 2–5x so với static batching cho cùng GPU (benchmark vLLM trên LLaMA-2-13B, A100).

Continuous batching chỉ áp dụng cho autoregressive model (decoder-only như GPT, LLaMA, Mistral). Encoder-only (BERT, bge) hay encoder-decoder (T5) dùng dynamic batching thông thường là đủ.

Các framework implement continuous batching:

• vLLM (PagedAttention + continuous batching)
• Text Generation Inference — TGI (Hugging Face)
• TensorRT-LLM (NVIDIA, in-flight batching)

vLLM (v0.6+) kết hợp PagedAttention (quản lý KV cache theo page, tránh fragmentation) với continuous batching. Đây là lý do throughput cao hơn so với naive HuggingFace Transformers pipeline.

Khởi động server:

pip install vllm

vllm serve meta-llama/Meta-Llama-3-8B-Instruct \
  --port 8000 \
  --max-num-seqs 256 \
  --max-model-len 4096

Server expose endpoint tương thích OpenAI API:

from openai import OpenAI

client = OpenAI(base_url="http://localhost:8000/v1", api_key="token")

response = client.chat.completions.create(
    model="meta-llama/Meta-Llama-3-8B-Instruct",
    messages=[{"role": "user", "content": "Explain batching in 2 sentences."}],
    max_tokens=100,
)
print(response.choices[0].message.content)

--max-num-seqs 256 là số sequence tối đa vLLM giữ đồng thời trong continuous batch. Tăng giá trị này tăng throughput nhưng tăng VRAM usage. Với A100 40GB + LLaMA-3 8B, 256 là điểm bắt đầu hợp lý.

Benchmark vLLM (arXiv:2309.06180, Kwon et al. 2023): throughput 24x so với HuggingFace Transformers naive serving trên LLaMA-13B.

NVIDIA Triton Inference Server hỗ trợ nhiều backend (PyTorch TorchScript, TensorRT, ONNX Runtime, TensorFlow) với dynamic batching cấu hình qua file config.pbtxt.

Cấu trúc model repository:

model_repository/
└── bge_small/
    ├── config.pbtxt
    └── 1/
        └── model.onnx

File config.pbtxt:

name: "bge_small"
backend: "onnxruntime"
max_batch_size: 64

input [
  { name: "input_ids"      data_type: TYPE_INT64  dims: [-1] },
  { name: "attention_mask" data_type: TYPE_INT64  dims: [-1] }
]
output [
  { name: "last_hidden_state" data_type: TYPE_FP32 dims: [-1, 384] }
]

dynamic_batching {
  max_queue_delay_microseconds: 20000   # 20ms
  preferred_batch_size: [16, 32]
}

instance_group [{ kind: KIND_GPU, count: 1 }]

Khởi động:

docker run --gpus all --rm \
  -v $(pwd)/model_repository:/models \
  -p 8000:8000 \
  nvcr.io/nvidia/tritonserver:24.05-py3 \
  tritonserver --model-repository=/models

Triton phù hợp khi cần serve nhiều loại model khác nhau (ONNX, TensorRT, PyTorch) trên cùng 1 server, hoặc khi cần ensemble pipeline (preprocessing → model → postprocessing). Với LLM autoregressive thuần túy, vLLM/TGI dễ setup hơn.

TorchServe (PyTorch official serving) có built-in batching qua file cấu hình config.properties:

batch_size=32
max_batch_delay=20

batch_size: số request tối đa per batch. max_batch_delay: thời gian chờ tối đa (ms).

Trong BaseHandler, method handle(data, context) nhận data là list có độ dài <= batch_size. Code handler xử lý như batch bình thường, không cần implement queue thủ công.

TorchServe thích hợp khi team đã dùng PyTorch và muốn serving solution chính thức, ít phụ thuộc ngoài. Triton linh hoạt hơn về backend nhưng setup phức tạp hơn.

Sentence-transformers hỗ trợ batching trực tiếp qua tham số batch_size của encode():

from sentence_transformers import SentenceTransformer

model = SentenceTransformer("BAAI/bge-small-en-v1.5")

texts = [...]  # danh sách N văn bản

# GPU: batch_size 32–64
embeddings = model.encode(
    texts,
    batch_size=32,
    normalize_embeddings=True,
    show_progress_bar=False,
)

# CPU: batch_size 8–16 là đủ, lớn hơn không tăng thêm nhiều
embeddings_cpu = model.encode(texts, batch_size=8, normalize_embeddings=True)

Một điểm cần lưu ý: các văn bản trong batch có độ dài khác nhau sẽ được padding đến độ dài max trong batch. Nếu có 1 văn bản dài bất thường, nó kéo toàn batch lên, gây waste compute. Giải pháp: sort texts theo độ dài trước khi encode, hoặc dùng convert_to_tensor=True với smart batching của sentence-transformers (tự nhóm theo length).

# Smart batching: sort theo length giảm padding waste
texts_sorted = sorted(texts, key=len)
embeddings = model.encode(texts_sorted, batch_size=32, normalize_embeddings=True)

Cả OpenAI và Anthropic cung cấp Batch API cho workload không cần realtime. Giá giảm 50% so với API thông thường, thời gian xử lý tối đa 24 giờ.

OpenAI Batch API

import json
import time
from openai import OpenAI

client = OpenAI()

# Bước 1: Tạo file JSONL
requests = [
    {
        "custom_id": f"req-{i}",
        "method": "POST",
        "url": "/v1/chat/completions",
        "body": {
            "model": "gpt-4o-mini",
            "messages": [{"role": "user", "content": f"Classify sentiment: {text}"}],
            "max_tokens": 10,
        },
    }
    for i, text in enumerate(texts)
]

with open("batch_input.jsonl", "w") as f:
    for req in requests:
        f.write(json.dumps(req) + "\n")

# Bước 2: Upload và tạo batch job
with open("batch_input.jsonl", "rb") as f:
    uploaded_file = client.files.create(file=f, purpose="batch")

batch_job = client.batches.create(
    input_file_id=uploaded_file.id,
    endpoint="/v1/chat/completions",
    completion_window="24h",
)

# Bước 3: Poll cho đến khi xong
while True:
    job = client.batches.retrieve(batch_job.id)
    if job.status in ("completed", "failed", "expired"):
        break
    time.sleep(60)

# Bước 4: Lấy kết quả
if job.status == "completed":
    result_content = client.files.content(job.output_file_id)
    for line in result_content.text.splitlines():
        result = json.loads(line)
        print(result["custom_id"], result["response"]["body"]["choices"][0]["message"]["content"])

Anthropic Batch API

Anthropic Messages Batch API (ra mắt 2024-10) có cấu trúc tương tự:

import anthropic
import time

client = anthropic.Anthropic()

# Tạo batch
batch = client.messages.batches.create(
    requests=[
        {
            "custom_id": f"req-{i}",
            "params": {
                "model": "claude-3-haiku-20240307",
                "max_tokens": 10,
                "messages": [{"role": "user", "content": f"Classify: {text}"}],
            },
        }
        for i, text in enumerate(texts)
    ]
)

# Poll
while True:
    status = client.messages.batches.retrieve(batch.id)
    if status.processing_status == "ended":
        break
    time.sleep(60)

# Lấy kết quả
for result in client.messages.batches.results(batch.id):
    if result.result.type == "succeeded":
        print(result.custom_id, result.result.message.content[0].text)

Use case phù hợp: large-scale labeling dataset, embedding toàn bộ document corpus, evaluation benchmark — bất kỳ task nào có thể chạy overnight mà không cần kết quả ngay.

Các metric cần theo dõi để đánh giá hiệu quả batching:

Ví dụ expose metrics với Prometheus trong DynamicBatcher:

from prometheus_client import Histogram, Gauge, Counter

batch_size_hist = Histogram("batcher_effective_batch_size", "Effective batch size", buckets=[1, 2, 4, 8, 16, 32, 64])
queue_depth_gauge = Gauge("batcher_queue_depth", "Current queue depth")
request_wait_hist = Histogram("batcher_request_wait_seconds", "Wait time before compute", buckets=[0.001, 0.005, 0.01, 0.02, 0.05, 0.1])

# Trong _worker, sau khi thu thập batch:
batch_size_hist.observe(len(batch))
queue_depth_gauge.set(self.queue.qsize())

OOM khi batch quá lớn

Batch lớn đẩy nhiều tensor lên GPU cùng lúc. Với model lớn (7B+), batch_size=32 có thể OOM trên GPU 24GB. Bắt đầu với batch_size nhỏ (4–8), tăng dần và monitor GPU memory qua nvidia-smi hoặc torch.cuda.memory_allocated().

max_wait_ms quá dài khi traffic thấp

Khi traffic thưa (dev, off-peak), request đến đơn lẻ sẽ luôn chờ đủ max_wait_ms trước khi được xử lý. Với max_wait_ms=50ms, mỗi request thêm 50ms latency không cần thiết. Xem xét adaptive wait hoặc đơn giản là giảm max_wait_ms xuống 5–10ms.

Không handle exception đúng cách

Nếu exception trong _process_batch không được catch và set vào từng future riêng, toàn bộ request trong batch sẽ treo (future never resolved). Luôn dùng try/except và fut.set_exception(exc) cho từng future:

except Exception as exc:
    for fut in futures:
        if not fut.done():
            fut.set_exception(exc)

Padding waste với input không đồng đều

Khi batch chứa text độ dài khác nhau (ví dụ 10 token và 512 token), tokenizer pad toàn batch lên 512 → 98% compute cho text ngắn là vô ích. Giải pháp: sort theo length trước khi batch, hoặc dùng bucket batching (nhóm text có độ dài tương đương vào cùng batch).

Lock contention trong high-throughput batcher

Dùng asyncio.Lock bao quanh mọi thao tác queue trong event loop single-thread thường không phải vấn đề, nhưng nếu batcher nhận input từ nhiều thread (non-async code), cần dùng loop.call_soon_threadsafe() thay vì lock thủ công. asyncio.Queue được khuyến khích vì đã thread-safe trong context asyncio.

Continuous batching không phù hợp mọi model

Continuous batching chỉ hoạt động với autoregressive decoder model. Encoder-only (BERT, bge) không có KV cache hay autoregressive decode step — không cần và không thể áp dụng. Dùng dynamic batching thông thường là đủ.

Bài 52: Rate Limiting — bảo vệ API khỏi abuse