xsl/hair
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(gateway): worker 池主备分组派发，主全挂才用备机

Browse Source 
- pool.py: WorkerState 增加 primary 标记；acquire_worker 分级派发——
  主组有在线只在主组内挑空闲（全忙排队，不退备组），主组全部下线才用备机
- config.py: 新增可选 primary_workers 配置（须为 workers 子集，自动过滤+告警）；
  不配置/空列表 = 全部视为主 worker，兼容旧行为
- config.json(不入 git): 3 台 GPU worker 设为主，max_global_concurrency 1→3
- app.py: /health 增加 workers_primary_total / workers_primary_healthy
- tests: 新增 4 条主备派发单测（主优先/主全挂用备/主全忙不走备/无配置兼容）
- docs(实现说明): 补充 primary_workers 配置说明，改配置需重启网关生效
This commit is contained in:
xsl
2026-09-22 23:48:21 +08:00
parent 6b24bf0b0d
commit f1b8b7bd82
6 changed files with 125 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gateway/pool.py
+27 -8
View File
@@ -46,6 +46,7 @@ class WorkerState:
url: str
online: bool = False # 初始 offline，等健康检查通过后上线
busy: bool = False
primary: bool = True # 主 worker（True 优先派发）；全部主 worker 下线才用备组
consecutive_failures: int = 0
consecutive_successes: int = 0
offline_until: float = 0.0 # 被动摘除后的隔离截止时刻（time.monotonic() 基准）
@@ -149,7 +150,12 @@ async def init_pool(cfg: dict) -> None:
"""初始化 worker 池并启动健康检查后台任务。"""
global _workers, _pool_condition, _health_task, _shutdown_event
_workers = {url: WorkerState(url=url) for url in cfg["workers"]}
# 主 worker 分组：配置了 primary_workers 则按其标记，否则全部视为主 worker
primary_set = set(cfg.get("primary_workers") or cfg["workers"])
_workers = {url: WorkerState(url=url, primary=url in primary_set)
for url in cfg["workers"]}
n_primary = sum(1 for w in _workers.values() if w.primary)
_pool_condition = asyncio.Condition()
_shutdown_event = asyncio.Event()
@@ -179,7 +185,8 @@ async def init_pool(cfg: dict) -> None:
# 对于 healthy_threshold == 1 的情况，已在上面的检查中上线
# 标记初始在线状态
online_count = sum(1 for w in _workers.values() if w.online)
logger.info("Worker 池初始化完成 | 总数=%d | 在线=%d", len(_workers), online_count)
logger.info("Worker 池初始化完成 | 总数=%d | 在线=%d | 主worker=%d",
len(_workers), online_count, n_primary)
# 启动后台健康检查
_health_task = asyncio.create_task(_health_check_loop(cfg))
@@ -273,6 +280,8 @@ async def mark_worker_unhealthy(w: WorkerState) -> None:
async def acquire_worker(cfg: dict) -> WorkerState:
"""从在线池中获取一个空闲 worker。
主备分组：只要还有主 worker（primary）在线，就只在主组内挑空闲
（全忙则排队等待，不用备组）；只有主组全部下线时才用备用 worker。
全忙则排队等待，最多 queue_wait_seconds；池空立即抛异常。
"""
queue_wait = cfg["dispatch"]["queue_wait_seconds"]
@@ -280,11 +289,13 @@ async def acquire_worker(cfg: dict) -> WorkerState:
async with _pool_condition:
while True:
# 在线 + 空闲
idle = [w for w in _workers.values() if w.online and not w.busy]
online = [w for w in _workers.values() if w.online]
primary_online = [w for w in online if w.primary]
# 主组有在线 → 只在主组内挑；主组全挂 → 才轮到备组
candidates = primary_online if primary_online else online
idle = [w for w in candidates if not w.busy]
if not idle:
online = [w for w in _workers.values() if w.online]
if not online:
raise NoWorkerAvailable("后端服务暂不可用，请稍后重试")
@@ -293,7 +304,8 @@ async def acquire_worker(cfg: dict) -> WorkerState:
if remaining <= 0:
raise NoWorkerAvailable("后端服务繁忙，请稍后重试")
logger.info("所有 worker 全忙（%d 在线），等待 %.1fs...", len(online), remaining)
logger.info("所有可用 worker 全忙（%d 在线），等待 %.1fs...",
len(candidates), remaining)
try:
await asyncio.wait_for(_pool_condition.wait(), timeout=remaining)
except asyncio.TimeoutError:
@@ -303,6 +315,8 @@ async def acquire_worker(cfg: dict) -> WorkerState:
# 取第一个空闲 worker
w = idle[0]
w.busy = True
if not w.primary:
logger.info("主 worker 全部下线，使用备用 worker: %s", w.url)
logger.debug("Worker 分配: %s", w.url)
return w
@@ -327,8 +341,13 @@ def get_pool_status() -> dict:
"global_waiting": _global_waiting,
}
if not _workers:
return {"total": 0, "healthy": 0, "busy": 0, **global_info}
return {"total": 0, "healthy": 0, "busy": 0,
"primary_total": 0, "primary_healthy": 0, **global_info}
total = len(_workers)
healthy = sum(1 for w in _workers.values() if w.online)
busy = sum(1 for w in _workers.values() if w.busy)
return {"total": total, "healthy": healthy, "busy": busy, **global_info}
primary_total = sum(1 for w in _workers.values() if w.primary)
primary_healthy = sum(1 for w in _workers.values() if w.primary and w.online)
return {"total": total, "healthy": healthy, "busy": busy,
"primary_total": primary_total, "primary_healthy": primary_healthy,
**global_info}