并发模型从「每worker并发1 + 多worker并行」改为全局串行:同一时间 只处理1个请求,其余排队;多 worker 仅作热备(主 worker 坏了才用备机)。 接口4(face/features) 走豆包、不占 GPU,不纳入串行。 - pool: asyncio.Semaphore(max_global_concurrency=1) + acquire/release_global_slot (依赖单进程 uvicorn 部署,已在注释中标注) - forward: proxy_request 最外层 acquire 全局槽、try/finally 全路径释放; 入参 multipart 解析挂 request.state;原重试/故障转移逻辑抽到 _dispatch_with_retries - reqlog(新): 标量入参保留;图片(file/base64)存盘转URL,绝不内嵌base64; 出参递归摘要截断(landmarks/长串/大数组) - logging_middleware: RequestLogEntry 加 request_params/response_data, jsonl 全量记录;_load_from_logfile 同步映射防重启丢字段;get_stats recent 暴露 - /gateway-health 暴露 global_max/global_busy/global_waiting - config: dispatch 新增 max_global_concurrency / max_queue_wait_seconds - tests: test_reqlog + test_gateway_serialization(9 用例) 顺带提交此前未提交的网关统计(daily stats 接口)与耗时看板(api_timing_dashboard.html)。 Co-Authored-By: Claude <noreply@anthropic.com>
59 lines
1.8 KiB
Python
59 lines
1.8 KiB
Python
"""gateway.pool 全局串行槽单测:验证 max_global_concurrency=1 时同一时间只跑1个请求、
|
|
排队超时返回 False。
|
|
|
|
用 asyncio.run 套同步测试,避免引入 pytest-asyncio 依赖。
|
|
"""
|
|
|
|
import asyncio
|
|
|
|
from gateway import pool
|
|
|
|
_CFG = {"dispatch": {"max_global_concurrency": 1, "max_queue_wait_seconds": 590}}
|
|
_CFG_FAST_TIMEOUT = {"dispatch": {"max_global_concurrency": 1, "max_queue_wait_seconds": 0.2}}
|
|
|
|
|
|
def test_global_slot_serializes_to_one():
|
|
pool.init_global_slot(_CFG)
|
|
|
|
state = {"active": 0, "peak": 0}
|
|
|
|
async def task():
|
|
got = await pool.acquire_global_slot(_CFG)
|
|
assert got is True
|
|
state["active"] += 1
|
|
state["peak"] = max(state["peak"], state["active"])
|
|
await asyncio.sleep(0.03)
|
|
state["active"] -= 1
|
|
pool.release_global_slot()
|
|
|
|
async def run():
|
|
await asyncio.gather(*(task() for _ in range(4)))
|
|
|
|
asyncio.run(run())
|
|
assert state["peak"] == 1, f"期望峰值并发=1,实际={state['peak']}"
|
|
assert pool.get_pool_status()["global_busy"] == 0
|
|
|
|
|
|
def test_global_slot_timeout_returns_false():
|
|
pool.init_global_slot(_CFG_FAST_TIMEOUT)
|
|
|
|
async def run():
|
|
first = await pool.acquire_global_slot(_CFG_FAST_TIMEOUT)
|
|
assert first is True
|
|
# 槽已被占用,第二个应在 0.2s 内超时
|
|
second = await pool.acquire_global_slot(_CFG_FAST_TIMEOUT)
|
|
assert second is False
|
|
pool.release_global_slot()
|
|
|
|
asyncio.run(run())
|
|
assert pool.get_pool_status()["global_busy"] == 0
|
|
|
|
|
|
def test_get_pool_status_has_global_fields():
|
|
pool.init_global_slot(_CFG)
|
|
status = pool.get_pool_status()
|
|
assert "global_max" in status
|
|
assert "global_busy" in status
|
|
assert "global_waiting" in status
|
|
assert status["global_max"] == 1
|