fix(tests): replace ThreadPoolExecutor with threading.Thread in RLock concurrency steps
ThreadPoolExecutor holds internal locks (e.g. _global_shutdown_lock) that can be in a locked state after fork(). The behave-parallel runner uses multiprocessing.Pool with the fork start method, so forked worker processes inherit these locked states and deadlock when trying to create or shut down a ThreadPoolExecutor. Replace both concurrent step implementations with direct threading.Thread usage, following the pattern established in context_tier_thread_safety_steps.py and other step files throughout the codebase. This eliminates the fork+lock deadlock while preserving the deterministic concurrency-counter approach for proving the RLock is released during transport calls.
This commit is contained in:
@@ -11,7 +11,6 @@ from __future__ import annotations
|
||||
|
||||
import threading
|
||||
import time
|
||||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||||
from typing import Any
|
||||
|
||||
from behave import given, then, when
|
||||
@@ -261,34 +260,60 @@ def step_lock_checking_discovery_transport(context: Context) -> None:
|
||||
|
||||
@when('I invoke "{tool_name}" concurrently from {count:d} threads')
|
||||
def step_concurrent_invoke(context: Context, tool_name: str, count: int) -> None:
|
||||
results: list[MCPToolResult] = []
|
||||
# Use threading.Thread directly instead of ThreadPoolExecutor to avoid
|
||||
# deadlocks in forked worker processes (behave-parallel uses fork).
|
||||
# ThreadPoolExecutor holds internal locks that can be in a locked state
|
||||
# after fork(), causing child processes to deadlock on acquisition.
|
||||
results: list[MCPToolResult | None] = [None] * count
|
||||
errors: list[BaseException | None] = [None] * count
|
||||
errors_lock = threading.Lock()
|
||||
|
||||
with ThreadPoolExecutor(max_workers=count) as executor:
|
||||
futures = [
|
||||
executor.submit(context.mcp_adapter.invoke, tool_name, {})
|
||||
for _ in range(count)
|
||||
]
|
||||
for future in as_completed(futures):
|
||||
results.append(future.result())
|
||||
def _invoke(idx: int) -> None:
|
||||
try:
|
||||
results[idx] = context.mcp_adapter.invoke(tool_name, {})
|
||||
except BaseException as exc:
|
||||
with errors_lock:
|
||||
errors[idx] = exc
|
||||
|
||||
context.concurrent_invoke_results = results
|
||||
threads = [threading.Thread(target=_invoke, args=(i,)) for i in range(count)]
|
||||
for t in threads:
|
||||
t.start()
|
||||
for t in threads:
|
||||
t.join()
|
||||
|
||||
# Re-raise the first error encountered, if any
|
||||
for err in errors:
|
||||
if err is not None:
|
||||
raise err
|
||||
|
||||
context.concurrent_invoke_results = [r for r in results if r is not None]
|
||||
|
||||
|
||||
@when("I call discover_tools concurrently from {count:d} threads")
|
||||
def step_concurrent_discover(context: Context, count: int) -> None:
|
||||
# Use threading.Thread directly instead of ThreadPoolExecutor to avoid
|
||||
# deadlocks in forked worker processes (behave-parallel uses fork).
|
||||
# ThreadPoolExecutor holds internal locks that can be in a locked state
|
||||
# after fork(), causing child processes to deadlock on acquisition.
|
||||
results: list[Any] = []
|
||||
errors: list[Exception] = []
|
||||
collect_lock = threading.Lock()
|
||||
|
||||
with ThreadPoolExecutor(max_workers=count) as executor:
|
||||
futures = [
|
||||
executor.submit(context.mcp_adapter.discover_tools) for _ in range(count)
|
||||
]
|
||||
for future in as_completed(futures):
|
||||
try:
|
||||
results.append(future.result())
|
||||
except Exception as exc:
|
||||
def _discover() -> None:
|
||||
try:
|
||||
result = context.mcp_adapter.discover_tools()
|
||||
with collect_lock:
|
||||
results.append(result)
|
||||
except Exception as exc:
|
||||
with collect_lock:
|
||||
errors.append(exc)
|
||||
|
||||
threads = [threading.Thread(target=_discover) for _ in range(count)]
|
||||
for t in threads:
|
||||
t.start()
|
||||
for t in threads:
|
||||
t.join()
|
||||
|
||||
context.concurrent_discover_results = results
|
||||
context.concurrent_discover_errors = errors
|
||||
|
||||
|
||||
Reference in New Issue
Block a user