yumoqing/pipeline-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: poller dedup dispatched tasks + error handling to prevent lock pile-up

Browse Source
This commit is contained in:
ymq 2026-08-08 13:43:19 +08:00
parent 6b295fc8ca
commit eb889d8694
1 changed files with 27 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
pipeline_service/init.py
View File

@ -531,9 +531,10 @@ def load_pipeline_service():
# Background poller: auto-dispatch submitted role_tasks to role agents
async def _role_poller(app):
"""启动后台轮询器，spawn 后立即返回（不阻塞 aiohttp 启动）。"""
"启动后台轮询器，spawn 后立即返回（不阻塞 aiohttp 启动）。"
from sqlor.dbpools import DBPools
poll_db = DBPools()
_dispatched = set() # 防止重复分发
async def _poll_loop():
while True:
@ -548,10 +549,19 @@ def load_pipeline_service():
tid = getattr(rec, 'id', '')
pid = getattr(rec, 'tenant_id', '')
r = getattr(rec, 'role', '')
if tid and pid and r:
if tid and pid and r and tid not in _dispatched:
_dispatched.add(tid)
debug(f"agent_poller dispatching task={tid} role={r}")
asyncio.ensure_future(
role_agent_loop(pid, r, agent_id=f"poller-{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)
asyncio.ensure_future(_dispatch(tid, pid, r))
except Exception as e:
debug(f"agent_poller error: {e}")
await asyncio.sleep(10)
@ -566,6 +576,7 @@ def load_pipeline_service():
async def _pm_poller(app):
from sqlor.dbpools import DBPools as _DBP
pm_db = _DBP()
_pm_dispatched = set()
async def _pm_poll_loop():
while True:
@ -579,10 +590,19 @@ def load_pipeline_service():
for rec in (recs or []):
tid = getattr(rec, 'id', '')
pid = getattr(rec, 'tenant_id', '')
if tid and pid:
if tid and pid and tid not in _pm_dispatched:
_pm_dispatched.add(tid)
debug(f"pm_poller dispatching review task={tid}")
asyncio.ensure_future(
pm_review_run(pid, agent_id="pm-poller"))
async def _pm_dispatch(tid, pid):
try:
await pm_review_run(pid, agent_id="pm-poller")
except Exception as e:
debug(f"pm_poller dispatch error: task={tid} err={e}")
finally:
_pm_dispatched.discard(tid)
asyncio.ensure_future(_pm_dispatch(tid, pid))
except Exception as e:
debug(f"pm_poller error: {e}")
await asyncio.sleep(15)