Enforce event loop affinity for cache instances (#732)

Author: rodrigobnogueira

README.rst

@@ -104,6 +104,53 @@ The library supports explicit invalidation for specific function call by
 The method returns `True` if corresponding arguments set was cached already, `False`
 otherwise.
 
+Limitations
+-----------
+
+**Event Loop Affinity**: ``alru_cache`` enforces that a cache instance is used with only
+one event loop. If you attempt to use a cached function from a different event loop than
+where it was first called, a ``RuntimeError`` will be raised:
+
+.. code-block:: text
+
+    RuntimeError: alru_cache is not safe to use across event loops: this cache
+    instance was first used with a different event loop.
+    Use separate cache instances per event loop.
+
+For typical asyncio applications using a single event loop, this is automatic and requires
+no configuration. If your application uses multiple event loops, create separate cache
+instances per loop:
+
+.. code-block:: python
+
+    import threading
+
+    _local = threading.local()
+
+    def get_cached_fetcher():
+        if not hasattr(_local, 'fetcher'):
+            @alru_cache(maxsize=100)
+            async def fetch_data(key):
+                ...
+            _local.fetcher = fetch_data
+        return _local.fetcher
+
+You can also reuse the logic of an already decorated function in a new loop by accessing ``__wrapped__``:
+
+.. code-block:: python
+
+    @alru_cache(maxsize=32)
+    async def my_task(x):
+        ...
+
+    # In Loop 1:
+    # my_task() uses the default global cache instance
+
+    # In Loop 2 (or a new thread):
+    # Create a fresh cache instance for the same logic
+    cached_task_loop2 = alru_cache(maxsize=32)(my_task.__wrapped__)
+    await cached_task_loop2(x)
+
 Benchmarks
 ----------
 

async_lru/__init__.py

@@ -112,10 +112,12 @@ def __init__(
         self.__closed = False
         self.__hits = 0
         self.__misses = 0
+        self.__first_loop: Optional[asyncio.AbstractEventLoop] = None
 
     @property
     def __tasks(self) -> List["asyncio.Task[_R]"]:
-        # NOTE: I don't think we need to form a set first here but not too sure we want it for guarantees
+        # NOTE: I don't think we need to form a set first here but not
+        # too sure we want it for guarantees
         return list(
             {
                 cache_item.task
@@ -124,6 +126,16 @@ def __tasks(self) -> List["asyncio.Task[_R]"]:
             }
         )
 
+    def _check_loop(self, loop: asyncio.AbstractEventLoop) -> None:
+        if self.__first_loop is None:
+            self.__first_loop = loop
+        elif self.__first_loop is not loop:
+            raise RuntimeError(
+                "alru_cache is not safe to use across event loops: this cache "
+                "instance was first used with a different event loop. "
+                "Use separate cache instances per event loop."
+            )
+
     def cache_invalidate(self, /, *args: Hashable, **kwargs: Any) -> bool:
         key = _make_key(args, kwargs, self.__typed)
 
@@ -144,6 +156,8 @@ def cache_clear(self) -> None:
         self.__cache.clear()
 
     async def cache_close(self, *, wait: bool = False) -> None:
+        loop = asyncio.get_running_loop()
+        self._check_loop(loop)
         self.__closed = True
 
         tasks = self.__tasks
@@ -222,6 +236,7 @@ async def __call__(self, /, *fn_args: Any, **fn_kwargs: Any) -> _R:
             raise RuntimeError(f"alru_cache is closed for {self}")
 
         loop = asyncio.get_running_loop()
+        self._check_loop(loop)
 
         key = _make_key(fn_args, fn_kwargs, self.__typed)
 
@@ -341,7 +356,6 @@ def wrapper(fn: _CBP[_R]) -> _LRUCacheWrapper[_R]:
         if not inspect.iscoroutinefunction(origin):
             raise RuntimeError(f"Coroutine function is required, got {fn!r}")
 
-        # functools.partialmethod support
         if hasattr(fn, "_make_unbound_method"):
             fn = fn._make_unbound_method()