Skip to content

Commit 4abb776

Browse files
committed
fix(tracing): stop set_tracing_processor_configs deadlocking on its own lock
TracingProcessorManager.set_processor_configs takes self.lock and then calls add_processor_config for each config, which takes the same lock again. self.lock was a threading.Lock, which is not reentrant, so the first call to the exported set_tracing_processor_configs() blocked forever: on an ACP server it never finishes startup, on a worker it never reaches worker.run(). Use an RLock so the batch still registers under one lock acquisition and the nested add can re-enter it. add_processor_config is unchanged. The new test registers two configs from a thread and fails on the old lock (the thread is still blocked after 5 s); it passes with the RLock.
1 parent 6d68f3a commit 4abb776

2 files changed

Lines changed: 53 additions & 2 deletions

File tree

‎src/agentex/lib/core/tracing/tracing_processor_manager.py‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -4,7 +4,7 @@
44
import logging
55
import threading
66
from typing import TYPE_CHECKING
7-
from threading import Lock
7+
from threading import RLock
88

99
from agentex.lib.types.tracing import TracingProcessorConfig
1010
from agentex.lib.core.tracing.processors.sgp_tracing_processor import (
@@ -36,7 +36,7 @@ def __init__(self):
3636
# Cache for processors
3737
self.sync_processors: list[SyncTracingProcessor] = []
3838
self.async_processors: list[AsyncTracingProcessor] = []
39-
self.lock = Lock()
39+
self.lock = RLock()
4040
self._agentex_registered = False
4141

4242
def _ensure_agentex_registered(self):
Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,51 @@
1+
from __future__ import annotations
2+
3+
import threading
4+
from typing import Any, cast
5+
from dataclasses import dataclass
6+
7+
from agentex.lib.core.tracing.tracing_processor_manager import TracingProcessorManager
8+
9+
10+
@dataclass
11+
class _FakeConfig:
12+
type: str = "fake"
13+
14+
15+
class _FakeProcessor:
16+
def __init__(self, config: Any) -> None:
17+
self.config = config
18+
19+
20+
def _manager() -> TracingProcessorManager:
21+
manager = TracingProcessorManager()
22+
manager.sync_config_registry["fake"] = cast(Any, _FakeProcessor)
23+
manager.async_config_registry["fake"] = cast(Any, _FakeProcessor)
24+
return manager
25+
26+
27+
def _finishes(target: Any, timeout: float = 5.0) -> bool:
28+
thread = threading.Thread(target=target, daemon=True)
29+
thread.start()
30+
thread.join(timeout)
31+
return not thread.is_alive()
32+
33+
34+
def test_set_processor_configs_registers_every_config_without_deadlocking() -> None:
35+
manager = _manager()
36+
configs = [_FakeConfig(), _FakeConfig()]
37+
38+
finished = _finishes(lambda: manager.set_processor_configs(cast(Any, configs)))
39+
40+
assert finished, "set_processor_configs blocked on the manager's own lock"
41+
assert [cast(Any, p).config for p in manager.get_sync_processors()] == configs
42+
assert [cast(Any, p).config for p in manager.get_async_processors()] == configs
43+
44+
45+
def test_add_processor_config_still_registers_one_pair() -> None:
46+
manager = _manager()
47+
48+
manager.add_processor_config(cast(Any, _FakeConfig()))
49+
50+
assert len(manager.get_sync_processors()) == 1
51+
assert len(manager.get_async_processors()) == 1

0 commit comments

Comments
 (0)