File size: 2,317 Bytes
a200053 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 | """Thread-safe lazy singletons for plugin authors (stdlib-only).
The ``if _client is None: _client = Expensive()`` footgun: two threads pass the guard, both
build, the second leaks the first's connections. :func:`lazy_singleton` decorates a zero-arg
accessor; :class:`SingletonSlot` is the manual slot when the instance depends on an argument.
"""
from __future__ import annotations
import functools
import threading
from typing import Callable, Generic, Optional, TypeVar
__all__ = ["lazy_singleton", "SingletonSlot"]
T = TypeVar("T")
class SingletonSlot(Generic[T]):
"""Thread-safe lazy slot: caches the first successfully-built instance ("first config wins").
The factory runs at most once under concurrent first calls; if it raises, nothing is cached
and the next call retries. ``_slot.get(lambda: Honcho(**resolve(config)))``."""
__slots__ = ("_lock", "_value", "_set")
def __init__(self) -> None:
self._lock = threading.Lock()
self._value: Optional[T] = None
self._set = False
def get(self, factory: Callable[[], T]) -> T:
# Fast path without the lock: a bool + ref read is atomic under the GIL.
if self._set:
return self._value # type: ignore[return-value]
with self._lock:
if self._set:
return self._value # type: ignore[return-value]
value = factory()
self._value = value
self._set = True
return value
def peek(self) -> Optional[T]:
"""Return the cached instance without building it (None if unset)."""
return self._value if self._set else None
def reset(self) -> None:
"""Drop the cached instance so the next ``get()`` rebuilds it."""
with self._lock:
self._value = None
self._set = False
def lazy_singleton(factory: Callable[[], T]) -> Callable[[], T]:
"""Wrap a zero-argument factory into a thread-safe lazy singleton accessor (factory runs once
even under concurrent first calls; on raise the next call retries). ``.reset()`` drops it."""
slot: SingletonSlot[T] = SingletonSlot()
@functools.wraps(factory)
def accessor() -> T:
return slot.get(factory)
accessor.reset = slot.reset # type: ignore[attr-defined]
return accessor
|