UbuntuandClaude
bb9f55e93c
并发模型从「每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
|