mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-09-11 14:38:38 +00:00
2f8d1cfc21
* fix(subagents): harden background-task registry and capacity snapshot edge cases - execute_async drops the just-registered background entry when submitting to the isolated loop fails. The caller sees the exception and never polls, and cleanup_background_task refuses non-terminal entries, so the entry would otherwise stay as a PENDING zombie forever. - SubagentExecutionCapacity.snapshot derives queued from len(_waiters) instead of iterating it. snapshot is read from non-loop threads (e.g. configure_subagent_execution_capacity) while the loop thread mutates the deque, so iteration can raise 'deque mutated during iteration'. The raw length may count a waiter that just timed out but has not removed itself yet, which only makes the busy-check more conservative. * fix(subagents): close failure-path gaps around background submit Address both review findings on the background submission lifecycle: - execute_async() copies the isolated-loop context before registering the _background_tasks entry, so a context-copy failure (callback- manager copy or loop-bound handler filtering) can no longer strand a permanent PENDING entry the caller will never poll. - _submit_to_isolated_loop_in_context() resolves the loop before calling the coroutine factory. As direct run_coroutine_threadsafe arguments the coroutine was created first, so a loop-startup failure stranded a never-awaited coroutine (RuntimeWarning + retained captures until collection). Both call sites share the fix. New tests verified red on the previous implementation, green after: - context-copy failure leaves no registry residue - the real submit helper (only the loop getter patched) never invokes the coroutine factory when loop startup fails * fix(subagents): close the coroutine when scheduling rejects it run_coroutine_threadsafe can itself raise once the coroutine exists (e.g. the loop closes between the lookup and the internal call_soon_threadsafe). Wrap the call, close the rejected coroutine, and re-raise; a focused test patches only run_coroutine_threadsafe and asserts the created coroutine reaches CORO_CLOSED.
60 lines
1.9 KiB
Python
60 lines
1.9 KiB
Python
"""Cross-thread safety of the shared subagent execution capacity."""
|
|
|
|
import asyncio
|
|
from collections import deque
|
|
|
|
import pytest
|
|
|
|
from deerflow.config.subagent_runtime_config import SubagentRuntimeConfig
|
|
from deerflow.subagents.capacity import SubagentExecutionCapacity
|
|
|
|
|
|
class _NoIterationDeque(deque):
|
|
"""A deque whose iteration always fails, as if mutated concurrently."""
|
|
|
|
def __iter__(self):
|
|
raise RuntimeError("deque mutated during iteration")
|
|
|
|
|
|
def test_snapshot_does_not_iterate_waiters():
|
|
"""snapshot() must be readable from a non-loop thread.
|
|
|
|
``configure_subagent_execution_capacity`` reads a snapshot while the loop
|
|
thread owns the waiters deque; iterating it cross-thread can raise
|
|
``deque mutated during iteration``. The snapshot must therefore derive
|
|
``queued`` without iterating.
|
|
"""
|
|
capacity = SubagentExecutionCapacity(SubagentRuntimeConfig(max_running=2, max_queued=4))
|
|
capacity._waiters = _NoIterationDeque(range(2))
|
|
|
|
snapshot = capacity.snapshot()
|
|
|
|
assert snapshot.max_running == 2
|
|
assert snapshot.max_queued == 4
|
|
assert snapshot.queued == 2
|
|
assert snapshot.running == 0
|
|
assert snapshot.admission_policy == "queue"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_snapshot_reports_running_and_queued_waiters():
|
|
"""The derived queued count still reflects pending waiters."""
|
|
capacity = SubagentExecutionCapacity(SubagentRuntimeConfig(max_running=1, max_queued=4, queue_timeout_seconds=5))
|
|
|
|
async def hold_slot():
|
|
async with capacity.slot():
|
|
await asyncio.sleep(60)
|
|
|
|
holder = asyncio.create_task(hold_slot())
|
|
queued = asyncio.create_task(capacity._acquire())
|
|
try:
|
|
await asyncio.sleep(0.05)
|
|
|
|
snapshot = capacity.snapshot()
|
|
assert snapshot.running == 1
|
|
assert snapshot.queued == 1
|
|
finally:
|
|
queued.cancel()
|
|
holder.cancel()
|
|
await asyncio.gather(queued, holder, return_exceptions=True)
|