From 6aff165d099d3d792801b11e83b7f7a3471d704f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=90=89=E9=91=AB=E5=8D=87?= Date: Wed, 24 Jun 2026 15:57:04 +0800 Subject: [PATCH] =?UTF-8?q?fix(db):=20=E8=B7=AF=E7=94=B1=E8=B7=A8=20loop?= =?UTF-8?q?=20=E7=9A=84=20DB=20=E8=AE=BF=E9=97=AE=E5=9B=9E=E5=BC=95?= =?UTF-8?q?=E6=93=8E=E4=B8=BB=20loop,=E6=B6=88=E9=99=A4=20asyncmy=20"attac?= =?UTF-8?q?hed=20to=20a=20different=20loop"?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 会商/定时任务/沙箱工具等在 worker 线程的临时 loop 上用 asyncio.run 跑全局引擎连接池的 DB 操作,连接被绑定到随即关闭的临时 loop;池子回收/释放该连接时 asyncmy _terminate_graceful_close 在错误的 loop 上触发 future,持续刷 "got Future attached to a different loop"。 - engine.py: 新增 run_db_blocking(fire_and_forget_db_write 的带返回值版本)。离开主 loop 时用 run_coroutine_threadsafe 把协程送回主 loop 执行、阻塞调用线程拿结果;仅在无主 loop 或正同步处于主 loop 内(送回会死锁)时退回临时 loop。 - thread_paths / prompt / skill_tools: 三处同步→异步阻塞读桥(任务owner解析/常驻技能/ 技能可见性)改走 run_db_blocking,删除各自重复的 _run_coro_blocking。 - llm/tool metrics 中间件: fire-and-forget 写改走 fire_and_forget_db_write。 Co-Authored-By: Claude Opus 4.8 (1M context) --- .../deerflow/agents/lead_agent/prompt.py | 25 +--- .../middlewares/llm_metrics_middleware.py | 18 +-- .../middlewares/tool_metrics_middleware.py | 17 +-- .../harness/deerflow/persistence/engine.py | 129 +++++++++++++++++- .../harness/deerflow/runtime/thread_paths.py | 26 +--- .../deerflow/tools/builtins/skill_tools.py | 28 +--- 6 files changed, 147 insertions(+), 96 deletions(-) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/prompt.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/prompt.py index e74190ca..7f7fca5d 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/prompt.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/lead_agent/prompt.py @@ -86,27 +86,6 @@ def _drop_archived_skills(skills: list[Skill]) -> list[Skill]: return skills -def _run_coro_blocking(coro): - """Run an async coroutine to completion from sync code. - - Mirrors ``skill_tools._run_coro_blocking``: when already inside a running - event loop (e.g. prompt built during an async agent run), hand the coroutine - to a fresh loop in a worker thread; otherwise run it directly. - """ - try: - asyncio.get_running_loop() - in_loop = True - except RuntimeError: - in_loop = False - - if in_loop: - import concurrent.futures - - with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool: - return pool.submit(asyncio.run, coro).result() - return asyncio.run(coro) - - def _load_always_on_names() -> set[str]: """Names marked 常驻(免压缩), read from the **DB** (skills table). @@ -114,11 +93,11 @@ def _load_always_on_names() -> set[str]: simply treats every skill as compressible (safe default). """ try: - from deerflow.persistence.engine import get_session_factory + from deerflow.persistence.engine import get_session_factory, run_db_blocking from deerflow.persistence.skills import make_skill_store store = make_skill_store(get_session_factory()) - return set(_run_coro_blocking(store.list_always_on_names())) + return set(run_db_blocking(store.list_always_on_names())) except Exception: logger.debug("Failed to load always_on skill names from DB", exc_info=True) return set() diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/llm_metrics_middleware.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/llm_metrics_middleware.py index e48fadfc..b3b93184 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/llm_metrics_middleware.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/llm_metrics_middleware.py @@ -20,7 +20,6 @@ not the retry-wrapped successful response. from __future__ import annotations -import asyncio import logging import time import uuid @@ -38,7 +37,7 @@ from langchain.agents.middleware.types import ( from langchain_core.messages import AIMessage from langgraph.errors import GraphBubbleUp -from deerflow.persistence.engine import get_session_factory +from deerflow.persistence.engine import fire_and_forget_db_write, get_session_factory from deerflow.persistence.llm_metrics import LlmMetricRecord, make_llm_metrics_store from deerflow.persistence.llm_metrics.base import LlmMetricsStore from deerflow.runtime.user_context import get_effective_user_id @@ -280,16 +279,13 @@ class LlmMetricsMiddleware(AgentMiddleware[AgentState]): store = self._store() if store is None: return + # Route the best-effort write onto the engine's home loop. Doing + # ``asyncio.run(store.record(...))`` here (when the model call ran in a worker + # thread) would reuse the global engine's pooled connections on a fresh loop and + # raise "got Future attached to a different loop" — harmless (caught below) but + # log-spammy and pool-churning under concurrency. See ``fire_and_forget_db_write``. try: - try: - loop = asyncio.get_running_loop() - except RuntimeError: - loop = None - if loop is not None and loop.is_running(): - # Fire-and-forget — never let metric IO block the request path. - loop.create_task(store.record(record)) - else: - asyncio.run(store.record(record)) + fire_and_forget_db_write(store.record(record)) except Exception: logger.warning("LlmMetricsMiddleware: failed to flush metric %s", record.id, exc_info=True) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/tool_metrics_middleware.py b/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/tool_metrics_middleware.py index 04c71ef5..ac609ac5 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/tool_metrics_middleware.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/agents/middlewares/tool_metrics_middleware.py @@ -25,7 +25,6 @@ agent runtime keeps running. from __future__ import annotations -import asyncio import logging import re import time @@ -41,7 +40,7 @@ from langgraph.errors import GraphBubbleUp from langgraph.prebuilt.tool_node import ToolCallRequest from langgraph.types import Command -from deerflow.persistence.engine import get_session_factory +from deerflow.persistence.engine import fire_and_forget_db_write, get_session_factory from deerflow.persistence.tool_metrics import ToolMetricRecord, make_tool_metrics_store from deerflow.persistence.tool_metrics.base import ToolMetricsStore from deerflow.runtime.user_context import get_effective_user_id @@ -262,16 +261,12 @@ class ToolMetricsMiddleware(AgentMiddleware[AgentState]): store = self._store() if store is None: return + # Route the best-effort write onto the engine's home loop (a tool call that ran + # in a worker thread would otherwise do ``asyncio.run(store.record(...))`` and reuse + # the global engine's pooled connections cross-loop → "got Future attached to a + # different loop"). See ``fire_and_forget_db_write``. try: - try: - loop = asyncio.get_running_loop() - except RuntimeError: - loop = None - if loop is not None and loop.is_running(): - # Fire-and-forget — never let metric IO block the tool path. - loop.create_task(store.record(record)) - else: - asyncio.run(store.record(record)) + fire_and_forget_db_write(store.record(record)) except Exception: logger.warning("ToolMetricsMiddleware: failed to flush metric %s", record.id, exc_info=True) diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/engine.py b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/engine.py index 69963f25..a426077a 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/persistence/engine.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/persistence/engine.py @@ -32,6 +32,13 @@ logger = logging.getLogger(__name__) _engine: AsyncEngine | None = None _session_factory: async_sessionmaker[AsyncSession] | None = None +# The event loop on which the async engine was created. asyncmy/asyncpg connection +# pools bind their connections to the loop that opened them; awaiting a write on a +# *different* loop reuses those connections cross-loop and raises +# "got Future attached to a different loop". We capture the engine's home loop here so +# best-effort writes coming from worker threads / other loops can be routed back to it +# (see ``fire_and_forget_db_write``). +_engine_loop: asyncio.AbstractEventLoop | None = None def _add_runtime_vendor_path() -> None: @@ -293,12 +300,20 @@ async def init_engine( pool_timeout: Seconds to wait for a free pooled connection before erroring (postgres/mysql). sqlite_dir: Directory to create for SQLite (ensured to exist). """ - global _engine, _session_factory + global _engine, _session_factory, _engine_loop if backend == "memory": logger.info("Persistence backend=memory -- ORM engine not initialized") return + # Remember the loop that owns the engine's connection pool, so off-loop + # best-effort writes (metrics from worker threads, etc.) can be routed here + # instead of reusing pooled connections cross-loop. + try: + _engine_loop = asyncio.get_running_loop() + except RuntimeError: + _engine_loop = None + if backend == "postgres": try: import asyncpg # noqa: F401 @@ -459,11 +474,121 @@ def get_engine() -> AsyncEngine | None: return _engine +def get_engine_loop() -> asyncio.AbstractEventLoop | None: + """Return the event loop that owns the engine's connection pool (or None).""" + return _engine_loop + + +def fire_and_forget_db_write(coro) -> None: + """Schedule a best-effort DB write on the engine's home loop, from any loop/thread. + + The async engine's connection pool is bound to the loop that created it + (``init_engine``, normally the Gateway's main loop). Running a write on a + *different* loop — e.g. a model/tool call that LangGraph executed in a worker + thread, which then does ``asyncio.run(store.record(...))`` — reuses those pooled + connections cross-loop and raises ``got Future attached to a different loop``. + Under concurrency this happens constantly (many calls land in worker threads at + once), spamming the log and churning the pool; single-threaded it almost never does. + + This routes every such write back onto the engine's home loop: + - already on that loop -> ``create_task`` (in-loop fire-and-forget); + - on another loop/thread -> ``run_coroutine_threadsafe`` to the home loop. + + The write itself stays best-effort: if no engine loop is captured (memory backend + / not initialized) the coroutine is closed (avoids a "never awaited" warning) and + the call is a no-op. Never raises. + """ + loop = _engine_loop + if loop is None or loop.is_closed(): + close = getattr(coro, "close", None) + if callable(close): + close() + return + try: + running = asyncio.get_running_loop() + except RuntimeError: + running = None + try: + if running is loop: + loop.create_task(coro) + else: + asyncio.run_coroutine_threadsafe(coro, loop) + except Exception: # noqa: BLE001 — best-effort; scheduling must never break the caller + close = getattr(coro, "close", None) + if callable(close): + close() + logger.debug("fire_and_forget_db_write: failed to schedule DB write", exc_info=True) + + +def _run_coro_on_throwaway_loop(coro): + """Last-resort: run *coro* to completion on a fresh event loop. + + Only reached when there is no usable engine home loop, or when the caller is + *synchronously* inside the home loop (routing back would deadlock). This + touches the shared connection pool from a throwaway loop, so it can still leak + a dead-loop-bound connection — but the common worker-thread path no longer + gets here (see ``run_db_blocking``). + """ + try: + asyncio.get_running_loop() + in_loop = True + except RuntimeError: + in_loop = False + + if in_loop: + import concurrent.futures + + with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool: + return pool.submit(asyncio.run, coro).result() + return asyncio.run(coro) + + +def run_db_blocking(coro, *, timeout: float | None = 30.0): + """Run an awaited DB coroutine from sync code, safe w.r.t. the engine's home loop. + + The blocking-read sibling of :func:`fire_and_forget_db_write`: where that one + drops the result, this returns it — used by the sync→async bridges that must + read a row before proceeding (thread-path owner resolution, always-on skills, + skill visibility). + + The async engine's connection pool is bound to the loop that created it + (``init_engine``). Running a DB coroutine on a *throwaway* loop + (``asyncio.run``) checks out pooled connections cross-loop and **leaks** them: + the connection ends up bound to a loop that is then closed, and when the pool + later recycles/disposes it, asyncmy's ``_terminate_graceful_close`` fires on the + wrong loop and raises ``got Future <...> attached to a different loop``. Under + concurrency (scheduler / subagent / sandbox-tool worker threads each spinning a + fresh loop) this floods the logs. + + So when we are **off** the home loop — the common case, a worker thread with no + running loop — route the coroutine onto the home loop and block the *caller* + thread for its result. The pool is only ever touched on its own loop, so nothing + leaks. Only when there is no home loop, or we are synchronously inside it (where + blocking would deadlock), do we fall back to a throwaway loop. + + Never swallows the coroutine's own exception — the sync bridges wrap this call in + their own try/except and degrade gracefully. + """ + loop = _engine_loop + try: + running = asyncio.get_running_loop() + except RuntimeError: + running = None + if loop is not None and not loop.is_closed() and running is not loop: + # Off the home loop (worker thread, or a different loop): execute the coro + # on the engine's loop and block this thread for the result. asyncmy's + # pooled connections stay on their own loop, so none leak cross-loop. + return asyncio.run_coroutine_threadsafe(coro, loop).result(timeout) + # No home loop, or we're already synchronously inside it: last resort. + return _run_coro_on_throwaway_loop(coro) + + async def close_engine() -> None: """Dispose the engine, release all connections.""" - global _engine, _session_factory + global _engine, _session_factory, _engine_loop if _engine is not None: await _engine.dispose() logger.info("Persistence engine closed") _engine = None _session_factory = None + _engine_loop = None diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/thread_paths.py b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/thread_paths.py index 28358b5c..5dbc8565 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/runtime/thread_paths.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/runtime/thread_paths.py @@ -28,7 +28,6 @@ safely to the pre-fix semantics. from __future__ import annotations -import asyncio import logging import threading @@ -106,27 +105,6 @@ def _is_creator_shared_thread(metadata: dict) -> bool: return bool(metadata.get("taskId")) or metadata.get("thread_type") == "roundtable" -def _run_coro_blocking(coro): - """Run an async coroutine to completion from sync code. - - Mirrors ``lead_agent.prompt._run_coro_blocking``: when already inside a - running event loop, hand the coroutine to a fresh loop in a worker thread; - otherwise run it directly. - """ - try: - asyncio.get_running_loop() - in_loop = True - except RuntimeError: - in_loop = False - - if in_loop: - import concurrent.futures - - with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool: - return pool.submit(asyncio.run, coro).result() - return asyncio.run(coro) - - def resolve_path_user_id(thread_id: str | None) -> str: """Canonical user bucket for *thread_id*'s files (sync). @@ -140,7 +118,9 @@ def resolve_path_user_id(thread_id: str | None) -> str: if cached is not None: return cached try: - owner = _run_coro_blocking(_fetch_task_owner(thread_id)) + from deerflow.persistence.engine import run_db_blocking + + owner = run_db_blocking(_fetch_task_owner(thread_id)) except Exception: logger.debug("Failed to resolve task-shared owner for thread %s", thread_id, exc_info=True) owner = None diff --git a/offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/skill_tools.py b/offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/skill_tools.py index c4ebb28b..caea5afa 100644 --- a/offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/skill_tools.py +++ b/offline-backend-20260512/backend/packages/harness/deerflow/tools/builtins/skill_tools.py @@ -75,30 +75,6 @@ def _resolve_effective_user_id() -> str | None: return None -def _run_coro_blocking(coro): - """在同步工具里把一个协程跑到完成。 - - ``skill_list`` 是同步 ``@tool``:LangChain 通常在线程池里执行它(无运行中的 - 事件循环),但 curator 评审会在已有事件循环的线程里调用它。两种情况都要能 - 拿到 ``SkillStore`` 的异步查询结果 —— 检测到运行中的事件循环就转交独立线程 - 执行 ``asyncio.run``,否则直接跑。 - """ - import asyncio - - try: - asyncio.get_running_loop() - in_loop = True - except RuntimeError: - in_loop = False - - if in_loop: - import concurrent.futures - - with concurrent.futures.ThreadPoolExecutor(max_workers=1) as pool: - return pool.submit(asyncio.run, coro).result() - return asyncio.run(coro) - - def _load_owned_skill_names(user_id: str | None) -> set[str] | None: """返回 ``owner_user_id == user_id`` 的自定义技能名集合。 @@ -110,14 +86,14 @@ def _load_owned_skill_names(user_id: str | None) -> set[str] | None: if not user_id: return None try: - from deerflow.persistence.engine import get_session_factory + from deerflow.persistence.engine import get_session_factory, run_db_blocking from deerflow.persistence.skills import make_skill_store factory = get_session_factory() if factory is None: return None store = make_skill_store(factory) - records = _run_coro_blocking(store.list_visible(user_id)) + records = run_db_blocking(store.list_visible(user_id)) return {r["name"] for r in records if r.get("owner_user_id") == user_id} except Exception: logger.debug("Failed to load owned skill names for skill_list filter", exc_info=True)