wehub-resource-sync
294 lines
10 KiB
Python
294 lines
10 KiB
Python
"""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()
|