diff --git a/pipeline_service/init.py b/pipeline_service/init.py index fe2c469..cd77c55 100644 --- a/pipeline_service/init.py +++ b/pipeline_service/init.py @@ -610,28 +610,82 @@ def load_pipeline_service(): # 达到并发上限,本轮不派发(退出 context 后末尾 sleep 再 poll) return + def _deps_of(depends_on_raw): + """解析 depends_on(JSON 数组字符串或 list)→ 依赖任务 ID 列表。空/非法 → []""" + if not depends_on_raw: + return [] + try: + d = json.loads(depends_on_raw) if isinstance(depends_on_raw, str) else depends_on_raw + return [str(x) for x in d if x] if isinstance(d, list) else [] + except (json.JSONDecodeError, TypeError, ValueError): + return [] + + # 候选任务(含 depends_on,LIMIT 放大以容纳「依赖未满足排在前面」的僵尸任务) recs = await sor.sqlExe( - "SELECT id, tenant_id, role FROM pipeline_tasks " + "SELECT id, tenant_id, role, depends_on FROM pipeline_tasks " "WHERE state='submitted' AND pipeline_id='role_task' " - "AND claimed_by IS NULL ORDER BY created_at ASC LIMIT 20", + "AND claimed_by IS NULL ORDER BY created_at ASC LIMIT 200", {}) - for rec in (recs or [])[:avail]: + + # 依赖门控提前到候选筛选:批量查 depends_on 里所有依赖任务的状态, + # 排除依赖未满足的「僵尸 submitted」任务——否则它们占住队列头、饿死其他就绪任务 + # (一项目卡死 → 全平台瘫痪)。终态 = completed/approved,与 _task_deps_satisfied 一致。 + all_dep_ids = set() + for rec in (recs or []): + all_dep_ids.update(_deps_of(getattr(rec, 'depends_on', '') or '')) + dep_state = {} + if all_dep_ids: + dep_ids = ",".join("'" + x.replace("'", "''") + "'" for x in all_dep_ids) + dep_rows = await sor.sqlExe( + "SELECT id, state FROM pipeline_tasks WHERE id IN (" + dep_ids + ")", {}) + dep_state = {getattr(r_, 'id', ''): getattr(r_, 'state', '') for r_ in (dep_rows or [])} + await sor.sqlExe("COMMIT", {}) + + ready = [] + for rec in (recs or []): + deps = _deps_of(getattr(rec, 'depends_on', '') or '') + # 依赖 id 查不到(任务已删)→ 视为未满足,保守不放行 + if all(dep_state.get(d, '') in ('completed', 'approved') for d in deps): + ready.append(rec) + + # 项目轮询(round-robin):每个项目轮流取一个就绪任务,保证项目间公平, + # 避免单项目大量就绪任务占满全局并发名额、饿死其他项目。(dict 保持插入序) + by_project = {} + for rec in ready: + by_project.setdefault(getattr(rec, 'tenant_id', ''), []).append(rec) + selected = [] + while len(selected) < avail and by_project: + progressed = False + for pid in list(by_project.keys()): + if len(selected) >= avail: + break + tasks = by_project[pid] + if tasks: + selected.append(tasks.pop(0)) + progressed = True + if not tasks: + del by_project[pid] + if not progressed: + break + + for rec in selected: tid = getattr(rec, 'id', '') pid = getattr(rec, 'tenant_id', '') r = getattr(rec, 'role', '') - if tid and pid and r and tid not in _dispatched: - _dispatched.add(tid) - debug(f"agent_poller dispatching task={tid} role={r}") + if not (tid and pid and r) or tid in _dispatched: + continue + _dispatched.add(tid) + debug(f"agent_poller dispatching task={tid} role={r}") - async def _dispatch(tid, pid, r): - try: - await role_agent_loop(pid, r, agent_id=f"poller-{r}") - except Exception as e: - debug(f"agent_poller dispatch error: task={tid} err={e}") - finally: - _dispatched.discard(tid) + async def _dispatch(tid, pid, r): + try: + await role_agent_loop(pid, r, agent_id=f"poller-{r}") + except Exception as e: + debug(f"agent_poller dispatch error: task={tid} err={e}") + finally: + _dispatched.discard(tid) - asyncio.ensure_future(_dispatch(tid, pid, r)) + asyncio.ensure_future(_dispatch(tid, pid, r)) async def _poll_loop(): while True: