fix: Share one lock per module for threadsafe resolution

injection/_core/common/threading.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 from contextlib import AbstractContextManager, nullcontext
 from os import getenv
 from threading import RLock
@@ -8,6 +10,14 @@
 )
 
 
-def get_lock(threadsafe: bool | None = None) -> AbstractContextManager[Any]:
+def get_lock(
+    threadsafe: bool | None = None,
+    *,
+    reuse: RLock | None = None,
+) -> AbstractContextManager[Any]:
     threadsafe = _PYTHON_INJECTION_THREADSAFE if threadsafe is None else threadsafe
-    return RLock() if threadsafe else nullcontext()
+
+    if not threadsafe:
+        return nullcontext()
+
+    return reuse or RLock()

injection/_core/module.py

@@ -35,6 +35,7 @@
 )
 from inspect import signature as inspect_signature
 from logging import Logger, getLogger
+from threading import RLock
 from types import MethodType
 from typing import TYPE_CHECKING, Any, ClassVar, Literal, NamedTuple, Self, overload
 
@@ -180,6 +181,11 @@ class Module(EventListener, InjectionProvider):  # type: ignore[misc]
         init=False,
         repr=False,
     )
+    __lock: RLock = field(
+        default_factory=RLock,
+        init=False,
+        repr=False,
+    )
     __loggers: list[Logger] = field(
         default_factory=lambda: [getLogger("python-injection")],
         init=False,
@@ -350,7 +356,7 @@ def create_metadata[**P, T](
         /,
         threadsafe: bool | None = None,
     ) -> InjectMetadata[P, T]:
-        return InjectMetadata(wrapped, threadsafe).listen(self)
+        return InjectMetadata(wrapped, threadsafe, lock=self.__lock).listen(self)
 
     if TYPE_CHECKING:  # pragma: no cover
 
@@ -397,7 +403,7 @@ async def afind_instance[T](
         *,
         threadsafe: bool | None = None,
     ) -> T:
-        with get_lock(threadsafe):
+        with get_lock(threadsafe, reuse=self.__lock):
             injectable = self[cls]
             return await injectable.aget_instance()
 
@@ -407,7 +413,7 @@ def find_instance[T](
         *,
         threadsafe: bool | None = None,
     ) -> T:
-        with get_lock(threadsafe):
+        with get_lock(threadsafe, reuse=self.__lock):
             injectable = self[cls]
             return injectable.get_instance()
 
@@ -835,9 +841,11 @@ def __init__(
         wrapped: Callable[P, T],
         /,
         threadsafe: bool | None = None,
+        *,
+        lock: RLock | None = None,
     ) -> None:
         self.__dependencies = Dependencies.empty()
-        self.__lock = get_lock(threadsafe)
+        self.__lock = get_lock(threadsafe, reuse=lock)
         self.__owner = None
         self.__tasks = deque()
         self.__wrapped = wrapped

tests/_core/common/test_threading.py

@@ -0,0 +1,18 @@
+from contextlib import nullcontext
+from threading import RLock
+
+from injection._core.common.threading import get_lock
+
+
+def test_get_lock_with_threadsafe_and_reuse_returns_reused_lock() -> None:
+    lock = RLock()
+
+    assert get_lock(threadsafe=True, reuse=lock) is lock
+
+
+def test_get_lock_with_threadsafe_without_reuse_returns_new_lock() -> None:
+    assert get_lock(threadsafe=True) is not get_lock(threadsafe=True)
+
+
+def test_get_lock_without_threadsafe_returns_null_context() -> None:
+    assert isinstance(get_lock(threadsafe=False, reuse=RLock()), nullcontext)

tests/_core/test_module.py

@@ -1,5 +1,7 @@
 import logging
-from collections.abc import AsyncIterator, Iterator
+import time
+from collections.abc import AsyncIterator, Callable, Iterator
+from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass
 from typing import Annotated, Any, Optional, Union
 
@@ -45,6 +47,20 @@ class C(B): ...
 class _Event(Event): ...
 
 
+def _register_slow_singleton(module: Module) -> None:
+    @module.singleton
+    def recipe() -> A:
+        time.sleep(0.05)
+        return A()
+
+
+def _run_concurrently(*functions: Callable[[], A]) -> set[A]:
+    with ThreadPoolExecutor() as executor:
+        futures = [executor.submit(function) for function in functions * 4]
+
+    return {future.result() for future in futures}
+
+
 class TestModule:
     def test_getitem_with_registered_class_returns_injectable(
         self,
@@ -696,6 +712,28 @@ def method(self: Any = None) -> None:
             class Service:
                 service = method
 
+    def test_inject_with_threadsafe(self, module: Module) -> None:
+        _register_slow_singleton(module)
+
+        @module.inject(threadsafe=True)
+        def function(a: A) -> A:
+            return a
+
+        assert len(_run_concurrently(function, function)) == 1
+
+    def test_inject_with_threadsafe_in_several_functions(self, module: Module) -> None:
+        _register_slow_singleton(module)
+
+        @module.inject(threadsafe=True)
+        def first(a: A) -> A:
+            return a
+
+        @module.inject(threadsafe=True)
+        def second(a: A) -> A:
+            return a
+
+        assert len(_run_concurrently(first, second)) == 1
+
     async def test_make_async_factory_with_async_dependency_returns_factory(
         self,
         module: Module,
@@ -712,6 +750,17 @@ class Service:
 
         assert isinstance((await factory()).a, A)
 
+    def test_find_instance_with_threadsafe_returns_same_singleton(
+        self,
+        module: Module,
+    ) -> None:
+        _register_slow_singleton(module)
+
+        def find() -> A:
+            return module.find_instance(A, threadsafe=True)
+
+        assert len(_run_concurrently(find, find)) == 1
+
     def test_find_instance_with_async_recipe_raises_runtime_error(
         self,
         module: Module,