1. 从一次线上事故说起：为什么并发控制是GPU推理服务的生死线
去年冬天，我帮一个团队收拾过一个烂摊子。他们用 FastAPI 包了一个视觉推理模型，部署到一台单卡 A10 的机器上，接口压测的时候一切正常，单请求延迟 80ms 左右，QPS 跑到 12 也没崩。结果上线第二天，晚高峰流量一上来，服务直接雪崩——不是慢慢变慢，是那种一瞬间全部请求卡死、健康检查超时、容器被反复重启的雪崩。登录机器一看，nvidia-smi显示显存占用 100%，进程还在，但 CUDA 上下文已经处于一种半死不活的状态，日志里刷满了CUDA out of memory。
这个场景太典型了。FastAPI 本身是个异步框架，天生适合处理高并发 IO 场景，但 GPU 推理是典型的计算密集型 + 独占资源型任务，两者放在一起，如果中间没有一层靠谱的并发控制，结果就是显存被瞬间打爆。很多人第一次做 GPU 推理服务，脑子里想的都是"FastAPI 快、异步、能扛并发"，却忽略了一个根本事实：你的 GPU 只有那么多显存，它不会因为你的 Web 框架是异步的就变得能装下更多东西。
这篇内容就是围绕这个核心矛盾展开的。我会把 FastAPI 做 GPU 推理服务时，并发控制该怎么设计、显存为什么会溢出、信号量怎么用、队列怎么排、批处理怎么权衡、超时和降级怎么做，一层一层拆开讲清楚。适合已经写过 FastAPI 接口、手里有 GPU 想部署模型、但被并发问题折磨过的开发者。如果你还在用 Flask 同步阻塞那一套，也能从里面看到迁移到 FastAPI 后必须补上的那门课。
2. 显存溢出的真实成因：不是请求太多，是并发模型错了
2.1 显存到底被谁吃掉了
先建立一个基本认知。一个 GPU 推理服务运行起来，显存占用大致分四块：
- 模型权重：这是固定开销。一个 7B 的 FP16 模型大概占 14GB，YOLOv11 这种检测模型可能只有几十 MB 到几百 MB。
- CUDA 上下文与运行时：PyTorch 加载后，CUDA context 本身会占几百 MB，这部分很多人会忽略。
- 推理中间激活值：这是随 batch size 线性甚至超线性增长的部分，也是并发失控时爆炸的元凶。
- 缓存与碎片：PyTorch 的 caching allocator 会预留显存，碎片化严重时明明有空闲却分配不出来。
关键在第三块。假设你的模型单条推理需要 200MB 激活显存，batch size 为 1 时占用很小。但如果 20 个请求同时进来，每个请求各自触发一次前向传播，激活显存就是 20 份叠加，再叠加 PyTorch 的缓存策略，很容易就顶到天花板。
2.2 异步框架的"假并发"陷阱
FastAPI 的async def端点看起来能同时处理很多请求，但这里有个致命误解：异步只解决 IO 等待，不解决计算排队。当你写：
@app.post("/infer") async def infer(req: InferRequest): result = model(req.data) # 同步阻塞的 GPU 计算 return result这个model(req.data)是同步阻塞调用，它会卡住整个事件循环。也就是说，你以为的"并发 20 个请求"，实际上是 20 个请求在事件循环里排队，一个算完再算下一个。这种情况下显存反而不容易爆，但吞吐量惨不忍睹，延迟线性叠加。
真正危险的是另一种写法——用run_in_executor或者线程池把推理丢到后台线程：
@app.post("/infer") async def infer(req: InferRequest): loop = asyncio.get_event_loop() result = await loop.run_in_executor(executor, model, req.data) return result这时候多个线程会真正同时调用 GPU，每个线程各自持有激活显存，显存占用瞬间翻倍。线程池默认大小如果是 CPU 核数，比如 16，那理论上同时可能有 16 份推理在跑，显存直接爆掉。这就是"请求一多就显存溢出"最直接的成因。
2.3 一个容易被忽视的细节：CUDA 上下文与多线程
还有一个坑，PyTorch 在多线程环境下调用 CUDA，每个线程默认会绑定到当前设备的默认流。多个线程同时往默认流里塞 kernel，虽然 CUDA 会串行化执行，但显存分配是并发的。也就是说，线程 A 还在等 kernel 执行，线程 B 已经开始为下一次推理分配显存了，分配器看到的是"当前已分配 + 待分配"的总和，很容易触发 OOM。
所以并发控制的核心目标不是"限制请求数"，而是限制同时处于 GPU 计算状态的推理任务数，并且让这个数字和你的显存容量精确匹配。
3. 并发控制方案选型：信号量、队列还是批处理
3.1 三种主流方案的对比
在 FastAPI 里做 GPU 并发控制，常见有三条路，我先把它们的特性摆出来：
| 方案 | 核心机制 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|---|
| 信号量限流 | asyncio.Semaphore控制并发数 | 实现简单，延迟可控 | 请求多了会排队等待，无背压 | 并发数固定、延迟敏感 |
| 显式队列 | asyncio.Queue+ 消费者协程 | 有背压，可控制队列长度 | 实现复杂，需要处理超时 | 高吞吐、可接受排队 |
| 动态批处理 | 攒批后一次前向 | 吞吐量最高，显存利用率好 | 延迟增加，实现最复杂 | 离线/准实时、吞吐优先 |
我的经验是：大多数线上服务用信号量就够了，因为它的心智负担最小，出问题也好排查。队列和批处理是吞吐量上不去时的进阶手段，不要一上来就搞复杂。
3.2 为什么信号量是首选
信号量的本质是一个计数器，acquire时减一，release时加一，减到零就阻塞等待。把它套在 GPU 推理上，逻辑非常清晰：
- 信号量初始值 = 你的 GPU 能同时安全承载的推理任务数。
- 每个请求进来先
acquire，拿到令牌才能调 GPU。 - 推理结束
release，把令牌还回去。
这个模型的好处是显存占用有确定上界。假设单任务峰值显存 500MB，信号量设为 4，那最坏情况就是 2GB 激活显存，加上模型权重和上下文，只要不超过总显存就绝对安全。这个数字是可以算出来的，不是拍脑袋。
3.3 信号量初始值怎么定
这是最关键的参数，定错了要么浪费 GPU，要么还是会 OOM。我的做法是分三步：
- 测单任务峰值显存：用
torch.cuda.max_memory_allocated()在单请求下测出峰值，记为M_single。 - 算可用显存：总显存减去模型权重、CUDA 上下文、系统预留，记为
M_avail。系统预留我一般留 1GB 到 1.5GB，防止碎片和临时分配。 - 算并发数：
N = floor(M_avail / M_single)，然后再打个八折，因为并发时碎片会更严重。
举个例子，A10 有 24GB 显存，模型权重 6GB，CUDA 上下文 0.5GB，系统预留 1.5GB，那M_avail = 24 - 6 - 0.5 - 1.5 = 16GB。如果单任务峰值 3GB，N = floor(16/3) = 5，打八折取 4。这个 4 就是信号量的初始值。
注意：这个数字不是一劳永逸的。模型换了、输入尺寸变了、PyTorch 版本升级了，都要重新测。我见过有人换了输入分辨率后没重测，结果并发数还是老值，直接 OOM。
4. FastAPI 中落地信号量并发控制的完整实现
4.1 项目目录结构
先把结构理清楚，避免所有代码堆在一个main.py里。这是我常用的组织方式：
gpu-infer-service/ ├── app/ │ ├── __init__.py │ ├── main.py # FastAPI 入口 │ ├── config.py # 配置（并发数、超时等） │ ├── model_loader.py # 模型加载与单例管理 │ ├── inference.py # 推理核心逻辑 │ ├── concurrency.py # 并发控制（信号量/队列） │ └── schemas.py # Pydantic 请求响应模型 ├── requirements.txt └── run.sh这个结构的好处是并发控制逻辑独立成模块，将来从信号量换成队列，只改concurrency.py，业务代码不动。
4.2 模型单例与预热
模型必须全局只加载一次，而且要在服务启动时预热，否则第一个请求会特别慢，还容易触发显存分配的抖动。
# app/model_loader.py import torch from threading import Lock class ModelHolder: _instance = None _lock = Lock() def __new__(cls): if cls._instance is None: with cls._lock: if cls._instance is None: cls._instance = super().__new__(cls) cls._instance._initialized = False return cls._instance def load(self, model_path: str, device: str = "cuda"): if self._initialized: return self.device = torch.device(device) self.model = torch.load(model_path, map_location=self.device) self.model.eval() # 预热：用假数据跑一次，触发 CUDA 上下文和缓存分配 dummy = torch.randn(1, 3, 224, 224).to(self.device) with torch.no_grad(): self.model(dummy) torch.cuda.synchronize() self._initialized = True model_holder = ModelHolder()预热这一步很多人省掉，结果线上第一个请求延迟 3 秒，还伴随显存分配的峰值抖动。跑一次假数据，让 caching allocator 把该分配的块都分配好，后续请求就稳了。
4.3 信号量并发控制模块
这是核心。注意我用的是asyncio.Semaphore，不是threading.Semaphore，因为我们要在异步端点里await。
# app/concurrency.py import asyncio from contextlib import asynccontextmanager class GPUConcurrencyController: def __init__(self, max_concurrent: int, acquire_timeout: float = 30.0): self.semaphore = asyncio.Semaphore(max_concurrent) self.acquire_timeout = acquire_timeout self.max_concurrent = max_concurrent @asynccontextmanager async def acquire(self): try: await asyncio.wait_for( self.semaphore.acquire(), timeout=self.acquire_timeout ) except asyncio.TimeoutError: raise RuntimeError("GPU 繁忙，排队超时") try: yield finally: self.semaphore.release()这里有几个设计决策值得说：
- 为什么加超时：如果不加超时，请求会无限排队，客户端早就超时了，服务端还在傻等，白白占着连接。30 秒是我常用的值，具体看你的 SLA。
- 为什么用
asynccontextmanager：保证release一定被执行，即使推理过程中抛异常。用 try/finally 手写也行，但 context manager 更不容易漏。 - 为什么不用
threading.Semaphore：因为推理如果丢到线程池，信号量在异步侧控制就够了，线程池那边不需要再控一层，否则容易死锁。
4.4 推理端点接入
把信号量接进端点，同时把同步推理丢到线程池，避免阻塞事件循环：
# app/main.py import asyncio from concurrent.futures import ThreadPoolExecutor from fastapi import FastAPI, HTTPException from app.model_loader import model_holder from app.concurrency import GPUConcurrencyController from app.schemas import InferRequest, InferResponse import torch app = FastAPI() executor = ThreadPoolExecutor(max_workers=4) controller = GPUConcurrencyController(max_concurrent=4, acquire_timeout=30.0) @app.on_event("startup") async def startup(): model_holder.load("model.pt", device="cuda") def _run_inference(data): tensor = torch.tensor(data).to(model_holder.device) with torch.no_grad(): out = model_holder.model(tensor) return out.cpu().numpy().tolist() @app.post("/infer", response_model=InferResponse) async def infer(req: InferRequest): try: async with controller.acquire(): loop = asyncio.get_event_loop() result = await loop.run_in_executor(executor, _run_inference, req.data) return InferResponse(result=result) except RuntimeError as e: raise HTTPException(status_code=503, detail=str(e))注意executor的max_workers和信号量的max_concurrent要一致。如果线程池比信号量大，多出来的线程拿不到令牌会空转；如果线程池比信号量小，信号量还没满线程池就满了，并发上不去。两者对齐是最省心的。
4.5 配置外置
并发数、超时这些参数不要写死在代码里，放到配置里，方便不同环境调整：
# app/config.py from pydantic_settings import BaseSettings class Settings(BaseSettings): max_concurrent: int = 4 acquire_timeout: float = 30.0 executor_workers: int = 4 model_path: str = "model.pt" device: str = "cuda" class Config: env_prefix = "INFER_" settings = Settings()这样部署时通过环境变量INFER_MAX_CONCURRENT=6就能调，不用改代码重新打包。
5. 进阶：从信号量到动态批处理
5.1 什么时候该上批处理
信号量方案在并发数不高时很好用，但有个天花板：GPU 利用率上不去。因为每个请求单独跑一次前向，kernel 启动开销、显存读写开销都摊不开。当你的 QPS 要求很高，而单请求延迟又不是特别敏感时，动态批处理能把吞吐量提升 3 到 10 倍。
原理很简单：把短时间内到达的多个请求攒成一个 batch，一次前向算完，再拆分结果返回。GPU 最擅长的就是并行处理大 batch，batch size 从 1 提到 8，延迟可能只增加 50%，但吞吐量翻了 8 倍。
5.2 一个可用的批处理实现
# app/batching.py import asyncio from dataclasses import dataclass, field from typing import Any, List @dataclass class BatchItem: data: Any future: asyncio.Future class DynamicBatcher: def __init__(self, max_batch_size: int, max_wait_ms: float, infer_fn): self.max_batch_size = max_batch_size self.max_wait = max_wait_ms / 1000.0 self.infer_fn = infer_fn self.queue: List[BatchItem] = [] self._lock = asyncio.Lock() async def submit(self, data) -> Any: loop = asyncio.get_event_loop() future = loop.create_future() item = BatchItem(data=data, future=future) async with self._lock: self.queue.append(item) if len(self.queue) == 1: asyncio.create_task(self._process_loop()) return await future async def _process_loop(self): while True: await asyncio.sleep(self.max_wait) async with self._lock: if not self.queue: return batch = self.queue[: self.max_batch_size] self.queue = self.queue[self.max_batch_size:] try: datas = [item.data for item in batch] results = await asyncio.get_event_loop().run_in_executor( None, self.infer_fn, datas ) for item, res in zip(batch, results): if not item.future.done(): item.future.set_result(res) except Exception as e: for item in batch: if not item.future.done(): item.future.set_exception(e)这个实现有两个关键参数：max_batch_size和max_wait_ms。前者受显存限制，后者决定延迟上限。我一般设max_wait_ms=20，也就是最多等 20ms 攒批，对大多数业务来说这个延迟增加可以接受。
5.3 批处理的显存陷阱
批处理虽然吞吐高，但显存占用是按 batch size 增长的。batch size 从 1 提到 8，激活显存可能涨 6 到 8 倍。所以max_batch_size必须用和信号量一样的方法测出来，不能拍脑袋。
而且批处理有个隐患：长短请求混在一起。如果 batch 里有一个请求的输入特别大，整个 batch 的显存就按最大的那个算。解决办法是按输入尺寸分桶，把相近尺寸的请求攒到一起。这个实现复杂度就上去了，一般业务量没到那个级别，用信号量反而更省心。
6. 常见问题与排查技巧实录
6.1 问题速查表
| 现象 | 可能原因 | 排查方法 | 解决 |
|---|---|---|---|
| 请求一多就 OOM | 并发数超过显存承载 | nvidia-smi看峰值，测单任务显存 | 调小信号量初始值 |
| 服务卡死无响应 | 同步推理阻塞事件循环 | 看 CPU 是否 100%，日志是否停更 | 推理丢线程池 |
| 延迟随并发线性增长 | 信号量排队 | 看 acquire 等待时间 | 提高并发数或上批处理 |
| 显存缓慢增长不释放 | 缓存碎片或张量泄漏 | 定期打印memory_allocated | 定期empty_cache或重启 |
| 第一个请求特别慢 | 未预热 | 看首请求延迟 | 启动时预热 |
| 偶发 CUDA error | 多线程竞争默认流 | 看是否多线程调 GPU | 用信号量串行化 |
6.2 几个踩过的坑
坑一：torch.no_grad()忘了加。推理时如果不加no_grad，PyTorch 会构建计算图，显存占用翻好几倍，而且不会自动释放。这个错误极其常见，尤其是从训练代码改过来的时候。
坑二：张量没及时释放。推理完的中间张量如果被引用着，显存不会回收。我习惯在推理函数里用局部变量，返回前.cpu().numpy()转走，让 GPU 张量自然被 GC。
坑三：empty_cache用错地方。torch.cuda.empty_cache()会释放缓存，但频繁调用会拖慢性能，因为下次分配又要重新申请。我一般只在长时间空闲后或者显存告警时调一次，不在每个请求后调。
坑四：uvicorn 多 worker 导致模型加载多份。uvicorn main:app --workers 4会启动 4 个进程，每个进程各自加载一份模型，显存直接乘 4。GPU 推理服务必须用单 worker，并发靠信号量和线程池解决，不要靠多进程。
坑五：健康检查打爆服务。K8s 的 liveness probe 如果也走推理端点，会占用信号量令牌。健康检查要单独开一个轻量端点，只检查进程存活，不碰 GPU。
6.3 监控指标建议
光靠日志排查太被动，建议暴露几个关键指标：
gpu_concurrency_in_use：当前占用的信号量数gpu_acquire_wait_seconds：获取令牌的等待时间gpu_memory_allocated_bytes：当前显存占用inference_duration_seconds：推理耗时分布
这几个指标一上，显存溢出前你就能看到acquire_wait飙升、memory_allocated逼近上限，提前扩容或限流。
7. 我个人在实际操作中的几点体会
做了几个 GPU 推理服务之后，我最大的体会是：并发控制不是加个信号量就完事，它是一套和你的模型、硬件、业务 SLA 强绑定的参数体系。同一个信号量数值，换个模型就不适用，换个输入尺寸也不适用。所以每次上线前，我都会老老实实跑一遍压测，测出单任务峰值显存、测出不同并发下的延迟曲线，再定参数。
另一个体会是不要过早优化。很多人一上来就想搞动态批处理、搞流式推理管线，结果代码复杂度爆炸，出了问题根本查不动。先用信号量把服务跑稳，等 QPS 真的成为瓶颈了，再考虑批处理。信号量方案的代码量可能只有批处理的五分之一，但能解决 80% 的场景。
最后分享一个小技巧：在推理函数里加一行torch.cuda.synchronize()再返回，虽然会稍微增加延迟，但能让显存释放和错误暴露更及时，排查问题时特别有用。线上稳定后可以去掉，但调试阶段强烈建议留着。
