github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-07-24 23:17:51 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(sandbox): synchronize sandbox provider singleton lifecycle (+ concurrency regression tests) (#3730)

Browse Source 
* fix(sandbox): synchronize sandbox provider singleton lifecycle

get_sandbox_provider() used an unsynchronized check-then-create, so two OS
threads (e.g. the main event loop and the Feishu channel thread, which runs
its own loop) could double-initialize the provider. With AioSandboxProvider
the overwritten instance leaks its idle-checker thread, since the only code
that joins it (shutdown()) is reachable only through the reference that was
overwritten.

reset_sandbox_provider(), shutdown_sandbox_provider() and set_sandbox_provider()
also touched the global without a lock, so a reset/shutdown racing an in-flight
create could clear it mid-construction or tear down an instance another thread
was about to return.

Guard all four lifecycle sites with a single module-level threading.Lock and
use double-checked locking in the getter, mirroring get_memory_storage().

* test(sandbox): add concurrent regression tests for provider singleton

- 8 threads racing on cold start, synchronized with a threading.Barrier so
  the check-then-create race fires deterministically; asserts exactly one
  provider instance is created.
- reset racing concurrent gets: asserts every returned value is a fully
  constructed provider (never None / half-built).

* fix(sandbox): lock get read path, run provider callbacks outside the lock

Addresses the three review findings on #3730:

1. get_sandbox_provider()'s read+return ran outside _provider_lock, so a
   concurrent reset/shutdown/set could null or tear the global between the
   check and the return, handing callers None / a torn instance. The hot read
   and the install reconciliation now both happen under the lock.

2. The non-reentrant _provider_lock was held across plugin-supplied callbacks
   (resolve_class import + provider __init__ in get; provider.reset()/shutdown()
   in reset/shutdown). A custom provider that re-entered these lifecycle
   functions would self-deadlock, and a slow teardown blocked every concurrent
   get(). resolve_class + construction now run outside the lock; reset/shutdown
   detach the reference under the lock and invoke the callback outside it.

   Tradeoff: racing cold-start callers may each construct a candidate. Exactly
   one is installed and returned to everyone; the losers (e.g. an AioSandbox
   instance that already started an idle-checker thread) are shut down so they
   do not leak the orphan thread #3721 is about. set_sandbox_provider() documents
   that it replaces but does not shut down the prior instance.

3. test_reset_racing_get reset the singleton to None *before* the barrier, so
   the racing reset was a no-op and never exercised reset-of-a-live-provider.
   It now populates the singleton up front so the reset tears down a live
   instance while getters read it.

Tests: rename the cold-start test to assert "one installed singleton, observed
by all" (construction is no longer single under the new design); add
shutdown-vs-get, set-vs-get, and a losing-racer-shuts-down-its-orphan case.
All five pass; the existing sandbox/middleware/mounts/uploads suites are green.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Czile 2026-06-25 08:06:17 +08:00 committed by GitHub
parent f656440245
commit 30841c3b9f
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 353 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

71
backend/packages/harness/deerflow/sandbox/sandbox_provider.py
View File

@ -1,4 +1,5 @@
import asyncio
import threading
from abc import ABC, abstractmethod
from deerflow.config import get_app_config
@ -55,6 +56,21 @@ class SandboxProvider(ABC):
_default_sandbox_provider: SandboxProvider | None = None
# Guards every read and write of `_default_sandbox_provider`. The singleton is
# reachable from more than one OS thread (e.g. the main event loop and the Feishu
# channel thread, which runs its own loop), so a bare check-then-create can double
# initialize the provider, and an unsynchronized reset/shutdown racing a get can
# hand a caller `None` or a torn instance. Every access to the global below takes
# this lock, including the read+return in `get_sandbox_provider()`.
#
# The lock guards only the reference swap. Provider callbacks (`__init__`,
# `reset()`, `shutdown()`) and the dynamic import in `resolve_class()` run
# *outside* the lock: they are plugin-supplied (`config.sandbox.use` resolves to
# an arbitrary class) and may be slow or, worse, re-enter these lifecycle
# functions. Holding a non-reentrant `threading.Lock` across them would
# self-deadlock such a provider and would block every concurrent `get()` during a
# slow teardown. Keeping callbacks off the lock avoids both.
_provider_lock = threading.Lock()
def get_sandbox_provider(**kwargs) -> SandboxProvider:
@ -67,11 +83,33 @@ def get_sandbox_provider(**kwargs) -> SandboxProvider:
A sandbox provider instance.
"""
global _default_sandbox_provider
if _default_sandbox_provider is None:
config = get_app_config()
cls = resolve_class(config.sandbox.use, SandboxProvider)
_default_sandbox_provider = cls(**kwargs)
return _default_sandbox_provider
# Fast path: a single locked read so a concurrent reset/shutdown can't null
# the global between the check and the return.
with _provider_lock:
if _default_sandbox_provider is not None:
return _default_sandbox_provider
# Cold start. Resolve + construct outside the lock: the import and the
# provider constructor are plugin code and must not run under a non-reentrant
# lock. The construction may race another caller; we reconcile under the lock.
config = get_app_config()
cls = resolve_class(config.sandbox.use, SandboxProvider)
provider = cls(**kwargs)
with _provider_lock:
if _default_sandbox_provider is None:
_default_sandbox_provider = provider
return provider
# We lost the install race: another thread got there first. `winner` is
# read under the same lock, so it is always a live instance, never None.
winner = _default_sandbox_provider
# Discard the instance we just built (outside the lock). For providers with
# side-effectful constructors (e.g. AioSandboxProvider starts an idle-checker
# thread), this tears down the orphan so it does not leak — issue #3721.
if hasattr(provider, "shutdown"):
provider.shutdown()
return winner
def reset_sandbox_provider() -> None:
@ -90,9 +128,13 @@ def reset_sandbox_provider() -> None:
Use `shutdown_sandbox_provider()` for proper cleanup.
"""
global _default_sandbox_provider
if _default_sandbox_provider is not None:
_default_sandbox_provider.reset()
# Detach the reference under the lock, then run the provider's `reset()`
# callback outside it (see the `_provider_lock` note).
with _provider_lock:
provider = _default_sandbox_provider
_default_sandbox_provider = None
if provider is not None:
provider.reset()
def shutdown_sandbox_provider() -> None:
@ -103,10 +145,13 @@ def shutdown_sandbox_provider() -> None:
is shutting down or when you need to completely reset the sandbox system.
"""
global _default_sandbox_provider
if _default_sandbox_provider is not None:
if hasattr(_default_sandbox_provider, "shutdown"):
_default_sandbox_provider.shutdown()
# Detach the reference under the lock, then run the (potentially slow)
# `shutdown()` callback outside it (see the `_provider_lock` note).
with _provider_lock:
provider = _default_sandbox_provider
_default_sandbox_provider = None
if provider is not None and hasattr(provider, "shutdown"):
provider.shutdown()
def set_sandbox_provider(provider: SandboxProvider) -> None:
@ -114,8 +159,12 @@ def set_sandbox_provider(provider: SandboxProvider) -> None:
This allows injecting a custom or mock provider for testing purposes.
Note: any previously installed provider is replaced but not shut down; the
caller owns the lifecycle of the instance it is overwriting.
Args:
provider: The SandboxProvider instance to use.
"""
global _default_sandbox_provider
_default_sandbox_provider = provider
with _provider_lock:
_default_sandbox_provider = provider

293
backend/tests/test_sandbox_provider_lifecycle.py Normal file
View File

@ -0,0 +1,293 @@
"""Concurrency regression tests for the sandbox provider singleton lifecycle.
These guard the fix for the unsynchronized check-then-create in
``get_sandbox_provider`` and the unlocked ``reset``/``shutdown``/``set`` paths:
before the lock was added, concurrent cold-start callers could each construct a
separate provider and overwrite the global, and a ``reset``/``shutdown`` racing
a ``get`` could hand a caller ``None`` or a torn-down instance.
Each test resets the process-global singleton on entry and in a ``finally`` on
exit, so tests never leak a provider into one another.
"""
import threading
import time
import deerflow.sandbox.sandbox_provider as sandbox_provider
from deerflow.sandbox.sandbox import Sandbox
from deerflow.sandbox.sandbox_provider import SandboxProvider
class SlowSandboxProvider(SandboxProvider):
"""Provider whose constructor is slow, to widen the check-then-create gap."""
instances_created = 0
instances_lock = threading.Lock()
def __init__(self) -> None:
time.sleep(0.05)
with self.instances_lock:
type(self).instances_created += 1
def acquire(self, thread_id: str | None = None) -> str:
return "sandbox-id"
def get(self, sandbox_id: str) -> Sandbox | None:
return None
def release(self, sandbox_id: str) -> None:
pass
class ShutdownSandboxProvider(SlowSandboxProvider):
"""Provider that also exposes ``shutdown``/``reset``, to exercise the paths
that run a provider callback outside ``_provider_lock``.
Every constructed instance registers itself in ``registry`` so a test can
assert which instances were later torn down.
"""
registry: list["ShutdownSandboxProvider"] = []
registry_lock = threading.Lock()
def __init__(self) -> None:
super().__init__()
self.shutdown_calls = 0
self.reset_calls = 0
with self.registry_lock:
type(self).registry.append(self)
def shutdown(self) -> None:
# A non-trivial teardown: the fix runs this outside the lock, so a
# concurrent get() must not be blocked or torn by it.
time.sleep(0.02)
self.shutdown_calls += 1
def reset(self) -> None:
self.reset_calls += 1
class _SandboxConfig:
use = "SlowSandboxProvider"
class _AppConfig:
sandbox = _SandboxConfig()
def _patch_provider_resolution(monkeypatch, cls=SlowSandboxProvider) -> None:
monkeypatch.setattr(sandbox_provider, "get_app_config", lambda: _AppConfig())
monkeypatch.setattr(sandbox_provider, "resolve_class", lambda *args: cls)
def test_get_sandbox_provider_installs_one_singleton_under_concurrent_access(monkeypatch):
"""Eight threads racing on a cold start must all observe the *same* installed
instance.
Construction runs outside ``_provider_lock`` (so plugin ``__init__``/import
never runs under a non-reentrant lock), so racing callers may each build a
candidate; the contract is that exactly one is installed and every caller
sees it. The losers are torn down — see
``test_losing_cold_start_racer_shuts_down_its_orphan``.
"""
sandbox_provider.reset_sandbox_provider()
SlowSandboxProvider.instances_created = 0
_patch_provider_resolution(monkeypatch)
n_threads = 8
providers: list[SandboxProvider] = []
providers_lock = threading.Lock()
# Barrier makes all threads enter get_sandbox_provider() at the same moment,
# so the race is triggered deterministically rather than probabilistically.
barrier = threading.Barrier(n_threads)
def get_provider() -> None:
barrier.wait()
provider = sandbox_provider.get_sandbox_provider()
with providers_lock:
providers.append(provider)
threads = [threading.Thread(target=get_provider) for _ in range(n_threads)]
for thread in threads:
thread.start()
for thread in threads:
thread.join()
try:
# Every caller sees the one installed singleton, whichever candidate won.
assert len({id(provider) for provider in providers}) == 1
installed = sandbox_provider.get_sandbox_provider()
assert all(p is installed for p in providers)
finally:
sandbox_provider.reset_sandbox_provider()
def test_reset_racing_get_of_live_singleton_never_returns_none_or_torn(monkeypatch):
"""A reset racing concurrent gets of a *live* singleton must never hand back
``None`` or a half-built instance: every returned value is a real provider.
The singleton is populated *before* the barrier so the resetter tears down a
live instance while the getters read it — the interleaving that the unlocked
get-read path could turn into a ``None``/torn return.
"""
sandbox_provider.reset_sandbox_provider()
SlowSandboxProvider.instances_created = 0
_patch_provider_resolution(monkeypatch)
# Populate the singleton up front so the reset races a live instance.
sandbox_provider.get_sandbox_provider()
results: list[object] = []
results_lock = threading.Lock()
barrier = threading.Barrier(5)
def getter() -> None:
barrier.wait()
provider = sandbox_provider.get_sandbox_provider()
with results_lock:
results.append(provider)
def resetter() -> None:
barrier.wait()
sandbox_provider.reset_sandbox_provider()
threads = [threading.Thread(target=getter) for _ in range(4)]
threads.append(threading.Thread(target=resetter))
for thread in threads:
thread.start()
for thread in threads:
thread.join()
try:
# Whatever each getter saw — the original singleton or a freshly rebuilt
# one after the reset — it must be a real provider, never None and never
# a partially constructed object.
assert results, "every getter recorded a result"
assert all(isinstance(p, SlowSandboxProvider) for p in results)
finally:
sandbox_provider.reset_sandbox_provider()
def test_shutdown_racing_get_of_live_singleton_never_returns_none_or_torn(monkeypatch):
"""Same guarantee as the reset case, for ``shutdown_sandbox_provider()``.
Uses a provider with a real (non-trivial) ``shutdown()`` so the teardown
runs outside the lock while getters read the global concurrently.
"""
sandbox_provider.reset_sandbox_provider()
SlowSandboxProvider.instances_created = 0
_patch_provider_resolution(monkeypatch, cls=ShutdownSandboxProvider)
sandbox_provider.get_sandbox_provider() # live singleton before the race
results: list[object] = []
results_lock = threading.Lock()
barrier = threading.Barrier(5)
def getter() -> None:
barrier.wait()
provider = sandbox_provider.get_sandbox_provider()
with results_lock:
results.append(provider)
def shutter() -> None:
barrier.wait()
sandbox_provider.shutdown_sandbox_provider()
threads = [threading.Thread(target=getter) for _ in range(4)]
threads.append(threading.Thread(target=shutter))
for thread in threads:
thread.start()
for thread in threads:
thread.join()
try:
assert results
assert all(isinstance(p, ShutdownSandboxProvider) for p in results)
finally:
sandbox_provider.reset_sandbox_provider()
def test_set_racing_get_never_returns_none_or_torn(monkeypatch):
"""``set_sandbox_provider()`` racing concurrent gets must never expose a
``None`` global: every getter sees a fully constructed provider."""
sandbox_provider.reset_sandbox_provider()
SlowSandboxProvider.instances_created = 0
_patch_provider_resolution(monkeypatch)
sandbox_provider.get_sandbox_provider() # live singleton before the race
injected = SlowSandboxProvider()
results: list[object] = []
results_lock = threading.Lock()
barrier = threading.Barrier(5)
def getter() -> None:
barrier.wait()
provider = sandbox_provider.get_sandbox_provider()
with results_lock:
results.append(provider)
def setter() -> None:
barrier.wait()
sandbox_provider.set_sandbox_provider(injected)
threads = [threading.Thread(target=getter) for _ in range(4)]
threads.append(threading.Thread(target=setter))
for thread in threads:
thread.start()
for thread in threads:
thread.join()
try:
assert results
assert all(isinstance(p, SlowSandboxProvider) for p in results)
finally:
sandbox_provider.reset_sandbox_provider()
def test_losing_cold_start_racer_shuts_down_its_orphan(monkeypatch):
"""When two cold-start callers race, the loser must shut down the instance it
built so a side-effectful constructor (idle-checker thread, etc.) does not
leak — the core consequence in issue #3721.
With ``ShutdownSandboxProvider`` every constructed-but-discarded instance has
its ``shutdown()`` invoked, so exactly ``(constructed - 1)`` of them are torn
down (the single winner is kept).
"""
sandbox_provider.reset_sandbox_provider()
ShutdownSandboxProvider.instances_created = 0
ShutdownSandboxProvider.registry = []
_patch_provider_resolution(monkeypatch, cls=ShutdownSandboxProvider)
n_threads = 8
providers: list[ShutdownSandboxProvider] = []
providers_lock = threading.Lock()
barrier = threading.Barrier(n_threads)
def get_provider() -> None:
barrier.wait()
provider = sandbox_provider.get_sandbox_provider()
with providers_lock:
providers.append(provider)
threads = [threading.Thread(target=get_provider) for _ in range(n_threads)]
for thread in threads:
thread.start()
for thread in threads:
thread.join()
try:
winner = sandbox_provider.get_sandbox_provider()
# Exactly one instance is installed and returned to every caller.
assert len({id(p) for p in providers}) == 1
assert all(p is winner for p in providers)
# The winner is never torn down...
assert winner.shutdown_calls == 0
# ...and every loser that was constructed had shutdown() called on it
# exactly once.
losers = [inst for inst in ShutdownSandboxProvider.registry if inst is not winner]
assert len(losers) == ShutdownSandboxProvider.instances_created - 1
assert all(inst.shutdown_calls == 1 for inst in losers)
finally:
sandbox_provider.reset_sandbox_provider()