diff --git a/injection/_core/common/threading.py b/injection/_core/common/threading.py index 42cf16c..109e000 100644 --- a/injection/_core/common/threading.py +++ b/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() diff --git a/injection/_core/module.py b/injection/_core/module.py index cc205ec..36a9832 100644 --- a/injection/_core/module.py +++ b/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 diff --git a/tests/_core/common/test_threading.py b/tests/_core/common/test_threading.py new file mode 100644 index 0000000..0f61e2d --- /dev/null +++ b/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) diff --git a/tests/_core/test_module.py b/tests/_core/test_module.py index 96cf6ca..777ce5f 100644 --- a/tests/_core/test_module.py +++ b/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,