diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 45844ee..4209a72 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -25,13 +25,21 @@ jobs: - "3.12" - "3.13" - "3.14" + # Free-threaded (PEP 703) build. Runs the same suite with the + # GIL disabled; see docs/free-threading.md. + - "3.14t" experimental: [ false ] steps: - - uses: actions/checkout@v3 + - uses: actions/checkout@v4 with: fetch-depth: 0 - - uses: "actions/setup-python@v4" + # setup-python must be >= v5.3: that is the first release that + # understands the free-threaded "t" suffix. On v4, "3.14t" is + # looked up as a literal version against arch x64 rather than + # x64-freethreaded, and the job fails with "The version '3.14t' + # with architecture 'x64' was not found". + - uses: "actions/setup-python@v5" with: python-version: "${{ matrix.python-version }}" cache: "pip" diff --git a/docs/free-threading.md b/docs/free-threading.md new file mode 100644 index 0000000..5a448a3 --- /dev/null +++ b/docs/free-threading.md @@ -0,0 +1,320 @@ +# Free-threaded Python (PEP 703) support + +`mode` supports free-threaded ("no-GIL") CPython. This page records what +was wrong before that was true, how each defect was fixed, and how to +re-check the work. + +Everything here was measured on **CPython 3.14.0rc2 free-threading build** +(`python3.14t`, `sys._is_gil_enabled() == False`), with a GIL-enabled +CPython 3.14.0rc2 used as the control. The reproducers live in +`tests/freethreading/stress.py`; the regression tests that keep the fixes +honest live in `tests/functional/test_thread_safety.py` and run on every +leg of the CI matrix. + +The "before" numbers are races, so the failure *rates* move between runs — +they are representative single runs, not stable constants. On repeated runs +the free-threaded `cached_property` figure ranged from 104/300 to 164/300, +and the cold-import figure from 14/25 to 18/25. What did not move is which +side of each table failed. + +## Status + +`mode` is pure Python, so there was never anything to *port* — it installed, +imported and passed its test suite on a free-threaded interpreter from the +start. What free threading changed is that four latent thread-safety defects +stopped being theoretical. One of them crashed the interpreter. + +All four are fixed. + +| | Free-threaded (before) | Free-threaded (after) | GIL | +|---|---|---|---| +| `pip install mode-streaming` | works (`py3-none-any`) | works | works | +| Import every `mode` module | GIL stays disabled | GIL stays disabled | n/a | +| `pytest tests/unit tests/functional` | passes | passes | passes | +| `LRUCache` under 16 threads | **SIGSEGV** | clean | clean | +| `cached_property` under 16 threads | **duplicate objects** | one object | one object | +| concurrent cold `import mode` | fails 14/25 runs | 0/25 | 0/25 | +| `Signal` under 16 threads | raises 30/30 | 0/30 | 0/30 | +| `mode[uvloop]` | GIL stays disabled | GIL stays disabled | n/a | +| `mode[gevent]` | **GIL re-enabled** | **GIL re-enabled** | n/a | + +`mode[gevent]` is the one item that is not fixed, because it cannot be +fixed here — see below. + +## What already worked + +No packaging work was required. `mode` ships no C extensions, so the +existing `py3-none-any` wheel already installs and runs on `3.13t`/`3.14t`. +Importing every module in the package leaves the GIL disabled, and the core +dependencies (`colorlog`, `croniter`, `mypy_extensions`) are pure Python. + +These were stress-tested with 16 concurrent OS threads and found **safe** +as they stood: + +- `Service` subclass creation — `__init_subclass__` writing the shared + `cls._tasks` mapping (`mode/services.py`) +- `ServiceThread` start/stop from many threads concurrently +- `get_event_loop()` — the `threading.local` cache in `mode/utils/loops.py` + correctly gives each thread its own loop with no cross-thread leakage +- `Node`/beacon tree traversal concurrent with mutation +- `ManagedUserDict` / `FastUserDict` mutation +- `annotations()` / `eval_type()` +- `LocalStack` — already `ContextVar`-based, so correct by construction + +## The four defects, and their fixes + +### 1. `LRUCache` could segfault the interpreter + +**Was: critical. Free-threading-specific.** + +`LRUCache.data` was a `collections.OrderedDict` and `thread_safety` +defaulted to `False`, which made the mutex a `nullcontext`. So eviction in +`__setitem__` and iteration in `keys()` ran with no lock at all. + +Under the GIL this was benign: 0/20 stress trials raised. On `3.14t` the +same code first raised `RuntimeError: OrderedDict changed size during +iteration` and then **segfaulted** — 4 of 5 runs of a 60-trial loop exited +with SIGSEGV, and a 5th hung. + +The cause was `OrderedDict` itself. Repeating the identical concurrent +mutate-and-iterate loop against a bare container: + +| container | free-threaded 3.14t | +|---|---| +| `collections.OrderedDict` | SIGSEGV / SIGABRT, 3/3 runs | +| plain `dict` | survives, 3/3 runs | + +Free-threaded CPython gives plain `dict` per-object locking; `OrderedDict`'s +C implementation did not get the same treatment, so concurrent mutation +corrupts its internal linked list. + +**Fixed** in `mode/utils/collections.py` by: + +- Making the mutex mandatory on free-threaded builds. `thread_safety` + defaults to the new `mode.utils.collections.FREE_THREADED` flag, checked + at runtime rather than build time so `PYTHON_GIL=1` is respected, and + passing `thread_safety=False` on such a build now raises `ValueError` + rather than handing back a structure that can take the interpreter down. +- Snapshotting in `_keys`/`_values`/`_items` instead of holding the mutex + across `yield`. The old code kept the lock held for as long as the + *consumer* took to iterate — and forever if the consumer abandoned the + generator, since the lock was only released when the generator was + closed. That hazard was latent while the lock defaulted to off; turning + the lock on by default would have made it real. + +### Why not just swap `OrderedDict` for `dict`? + +That was the first fix, and it was wrong. `dict` has preserved insertion +order since 3.7 and is memory-safe under free threading, so it looks like a +free win — but `LRUCache`'s hot path is evicting the *oldest* entry, and +that is the one thing `dict` cannot do in O(1). `OrderedDict.popitem(last= +False)` unlinks a node; the `dict` equivalent, `d.pop(next(iter(d)))`, has +to scan past every slot vacated since the last resize. + +Steady-state evict-and-insert, 100k operations: + +| cache size | `OrderedDict` | `dict` | +|---|---|---| +| 1,000 | 0.043s | 0.089s | +| 10,000 | 0.046s | 0.448s | +| 100,000 | 0.052s | 2.447s | + +The gap grows linearly with the cache, because the eviction itself became +O(n). Periodically rebuilding the dict to compact it only softens this to +O(√n) — still ~24x at 100k — so there is no cheap repair. `OrderedDict` is +the right data structure here; the concurrency hazard belongs to the mutex, +not to the choice of container. + +`LRUCache` is not used inside `mode` itself; it is exported utility surface +(faust is a consumer), so the blast radius was downstream. + +### 2. `cached_property` handed different objects to different threads + +**Was: high. Free-threading-specific.** + +`cached_property.__get__` was a check-then-act on `obj.__dict__`: try the +key, catch `KeyError`, compute, store. Nothing made that atomic. + +| | duplicate-object trials | computes per 300 properties | +|---|---|---| +| GIL 3.14 | 0/300 | 300 | +| free-threaded 3.14t | **104/300** | 419 | + +This was not merely wasted work. `ServiceProxy` documents +`@cached_property _service` as *the* way to build the proxied service — it +is how the Faust App is constructed at module level. Racing 16 threads on +`proxy._service`: + +| | trials that built/returned >1 `Service` | +|---|---| +| GIL 3.14 | 0/200 | +| free-threaded 3.14t | **198/200** | + +So one thread could `start()` one `Service` instance while another held a +different instance, and the later `stop()` never reached the one that was +started. + +**Fixed** in `mode/utils/objects.py` with double-checked locking: the +already-cached lookup stays lock-free (a plain dict hit), and only the miss +path takes a per-descriptor `RLock` and re-checks after acquiring. +Contention is therefore limited to first-time initialisation. + +Note that stdlib `functools.cached_property` deliberately dropped its lock +in 3.12 and accepts duplicate computation. That trade-off is fine for a +pure value cache; it is not fine for a singleton service handle. + +### 3. Concurrent first `import mode` could hand back a half-built module + +**Was: high. Pre-existing, but much worse under free threading.** This one +broke the most ordinary thing a user does. + +`mode/__init__.py` used the Werkzeug lazy-import trick: define a `_module` +subclass whose `__getattr__` resolves the lazily-exported names, then swap +it into `sys.modules` at the *end* of the module body. + +If thread B ran `import mode` while thread A was still executing +`mode/__init__.py`, B could be handed the original, pre-swap module object — +which has no `__getattr__` — so every lazily-exported name raised: + +``` +AttributeError: module 'mode' has no attribute 'Service' +``` + +Racing 16 threads on a cold `import mode` followed by attribute access: + +| | runs with at least one failing thread | +|---|---| +| GIL 3.14 | 3/25 | +| free-threaded 3.14t | **14/25** | + +Instrumenting a failing thread confirmed the mechanism: the object it +imported was a plain `module` while `sys.modules["mode"]` was already the +`_module` instance — the thread held the stale pre-swap object. The +replacement module also carried **no `__spec__`**, which deprived the import +machinery of the `_initializing` flag it would otherwise use to make the +second thread wait. + +**Fixed** by dropping the `sys.modules` swap entirely in favour of a +:pep:`562` module-level `__getattr__` (plus a module `__dir__`). PEP 562 +landed in 3.7 and mode's floor is 3.10, so the `_module` class existed only +for compatibility that is no longer needed. With no swap, the race cannot +happen — and `sys.modules["mode"]` keeps its real `__spec__`. + +### 4. `Signal` mutated its receiver set during iteration + +**Was: medium. Pre-existing, not a free-threading regression** — it raised +`RuntimeError: Set changed size during iteration` in 30/30 trials on *both* +builds, so `Signal` had never been thread-safe. + +`_get_live_receivers` iterated `self._receivers` (a plain `set`) while +`connect`/`disconnect` added to and discarded from it — and the caller then +discarded dead refs from the same set using the result. + +**Fixed** in `mode/signals.py` by iterating a snapshot. + +The snapshot must be `list(r)`, **not** `tuple(r)`. This is not stylistic: + +| snapshot of a set being mutated by 4 threads | free-threaded 3.14t | +|---|---| +| `tuple(s)` | **8 failures** — `Set changed size during iteration` | +| `list(s)` | 0 failures | +| `set(s)` | 0 failures | +| `s.copy()` | 0 failures | +| `frozenset(s)` | 0 failures | + +`list()`, `set()` and `set.copy()` take the source set's per-object lock for +the duration of the copy; `tuple()` falls back to the generic iterator +protocol and does not, so `tuple(r)` raises the very error the snapshot +exists to prevent. The first attempt at this fix used `tuple(r)` and the +stress harness caught it. + +### Not fixable here: the `gevent` extra re-enables the GIL + +| extra | result on `3.14t` | +|---|---| +| `mode[uvloop]` | uvloop 0.22.1 imports and runs, GIL stays disabled | +| `mode[eventlet]` | imports, GIL stays disabled (eventlet prints its own migrate-away notice) | +| `mode[gevent]` | **GIL re-enabled at import** | + +Installing `mode[gevent]` silently downgrades a free-threaded interpreter +back to GIL semantics: + +``` +RuntimeWarning: The global interpreter lock (GIL) has been enabled to load +module 'gevent.libev.corecext', which has not declared that it can run +safely without the GIL. +``` + +This is upstream in gevent, not something `mode` can fix. It is flagged in +`pyproject.toml` next to the extra, and `mode/loop/gevent.py` now warns at +import time on a free-threaded build — the degradation is otherwise silent, +since you keep running and simply are not free-threaded any more. That check +uses the *build* flag (`sysconfig.get_config_var("Py_GIL_DISABLED")`) rather +than `sys._is_gil_enabled()`, which by then already reads `True`. + +**Separately: `mode.loop.use("gevent")` is currently broken on every build.** +This has nothing to do with free threading — it fails identically on +GIL-enabled 3.10 and 3.14 with gevent 26.7.0: + +``` +ImportError: Cannot import 'Loop' from +``` + +The cause is a self-referential import. `mode/loop/gevent.py` sets +`GEVENT_LOOP=mode.loop._gevent_loop.Loop`, but `mode/loop/_gevent_loop.py` +imports `gevent.core` at module scope in order to subclass +`gevent.core.loop`. Importing it therefore builds a gevent hub, which +resolves `GEVENT_CONFIG.loop`, which imports `mode.loop._gevent_loop` — a +module whose body has not yet reached `class Loop`. Pre-importing the module +does not help, because the cycle is inside its own import. + +gevent itself is fine: `gevent.monkey.patch_all()` plus +`asyncio_gevent.EventLoopPolicy` runs an asyncio coroutine correctly. Only +mode's custom `GEVENT_LOOP` hook fails. Presumably gevent used to resolve +that setting lazily and no longer does. + +`mode.loop` has no test coverage, which is how this went unnoticed. Fixing +it would mean building `Loop` lazily rather than at module scope. + +Rather than repair a backend that cannot work on free-threaded builds +anyway, the gevent loop is **deprecated**: selecting it raises a +`DeprecationWarning` naming the breakage and pointing at `aio`/`uvloop`, +and it is slated for removal in a future major release. Nothing is removed +yet, so this is not a breaking change. + +## CI + +`3.14t` is part of the `tests.yml` matrix, so the suite — including +`tests/functional/test_thread_safety.py` — runs with the GIL disabled on +every push. `ruff` and `mypy` both run clean on the free-threaded build. + +The package advertises +`Programming Language :: Python :: Free Threading :: 2 - Beta`. + +### A note on `pytest-run-parallel` + +`pytest-run-parallel` installs and runs on `3.14t`, but pointing +`--parallel-threads` at the existing suite is not useful: it reports ~33 +failures in `tests/functional/utils/test_collections.py` alone that are +artifacts of tests sharing mutable fixtures and `Mock` objects, not mode +bugs. For example `test_AttributeDictMixin::test_set_get` fails with "DID +NOT RAISE AttributeError" purely because a sibling thread already set the +attribute on the shared object. + +Use it selectively on purpose-written thread-safety tests rather than across +the whole suite. + +## Reproducing + +```sh +uv python install 3.14t +uv venv --python 3.14t .venv-ft +VIRTUAL_ENV=.venv-ft uv pip install -e . -r requirements-tests.txt +.venv-ft/bin/python -m pytest tests/unit tests/functional +.venv-ft/bin/python tests/freethreading/stress.py +``` + +`tests/freethreading/` is deliberately outside the `testpaths` configured in +`pyproject.toml`, so the heavier probabilistic reproducers are never +collected by a normal `pytest` run. Run the same file under a GIL-enabled +interpreter to see the control numbers. diff --git a/mkdocs.yml b/mkdocs.yml index 84d2b1f..23b43ca 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -39,6 +39,7 @@ nav: - Web app: example-webapp.md - Developing: - Contributing Guide: contributing.md + - Free-threaded Python: free-threading.md - References: - 'Mode': - mode.services: references/mode.services.md diff --git a/mode/__init__.py b/mode/__init__.py index eaca124..2f377c2 100644 --- a/mode/__init__.py +++ b/mode/__init__.py @@ -2,13 +2,11 @@ __version__ = "0.0.1" -import sys import typing from collections.abc import Mapping, Sequence -# Lazy loading. -# - See werkzeug/__init__.py for the rationale behind this. -from types import ModuleType +# Lazy loading, via the PEP 562 module __getattr__ defined at the bottom +# of this file. from typing import Any # -eof meta- @@ -88,47 +86,47 @@ object_origins[item] = module -class _module(ModuleType): - """Customized Python module.""" - - def __getattr__(self, name: str) -> Any: - if name in object_origins: - module = __import__(object_origins[name], None, None, [name]) - for extra_name in all_by_module[module.__name__]: - setattr(self, extra_name, getattr(module, extra_name)) - return getattr(module, name) - return ModuleType.__getattribute__(self, name) - - def __dir__(self) -> Sequence[str]: - result = list(new_module.__all__) - result.extend( - ( - "__file__", - "__path__", - "__doc__", - "__all__", - "__docformat__", - "__name__", - "__path__", - "VERSION", - "version_info", - "__package__", - ) - ) - return result - - -# keep a reference to this module so that it's not garbage collected -old_module = sys.modules[__name__] - -new_module = sys.modules[__name__] = _module(__name__) -new_module.__dict__.update( - { - "__file__": __file__, - "__path__": __path__, - "__doc__": __doc__, - "__all__": tuple(object_origins), - "__version__": __version__, - "__package__": __package__, - } -) +# NOTE: This is a :pep:`562` module-level ``__getattr__``, and deliberately +# *not* the older trick of defining a ``ModuleType`` subclass and swapping it +# into ``sys.modules[__name__]`` at the end of this file. +# +# That swap was a race: it only happened once the module body had finished, +# so a thread calling ``import mode`` while another thread was still +# executing this file could be handed the original, pre-swap module object -- +# which has no ``__getattr__`` on it -- and every lazily-exported name below +# raised ``AttributeError: module 'mode' has no attribute 'Service'``. The +# replacement module also carried no ``__spec__``, which denied the import +# machinery the ``_initializing`` flag it uses to make the second thread wait. +# Rare under the GIL, common on free-threaded (:pep:`703`) builds. +# +# A module ``__getattr__`` needs no swap at all, so the race cannot happen. +def __getattr__(name: str) -> Any: + try: + origin = object_origins[name] + except KeyError: + raise AttributeError( + f"module {__name__!r} has no attribute {name!r}" + ) from None + module = __import__(origin, None, None, [name]) + # Bind every name this module provides, not just the requested one, so + # that later lookups are plain globals and never reach __getattr__ again. + namespace = globals() + for extra_name in all_by_module[origin]: + namespace[extra_name] = getattr(module, extra_name) + return namespace[name] + + +def __dir__() -> Sequence[str]: + return [ + *__all__, + "__file__", + "__path__", + "__doc__", + "__all__", + "__docformat__", + "__name__", + "VERSION", + "version_info", + "__package__", + "__version__", + ] diff --git a/mode/locals.py b/mode/locals.py index 973d24c..919024f 100644 --- a/mode/locals.py +++ b/mode/locals.py @@ -152,6 +152,28 @@ class XProxy(MutableMappingRole, AsyncContextManagerRole): PYPY = hasattr(sys, "pypy_version_info") SLOTS_ISSUE_PRESENT = sys.version_info < (3, 7) + +def _cooperative_init_subclass(cls: "type[Proxy[Any]]") -> None: + """Call the next ``__init_subclass__`` in ``Proxy``'s MRO. + + This lives at module level, outside the class body, for one reason: + naming ``super`` inside a method of ``Proxy`` would make the compiler + add an implicit ``__class__`` closure cell to that class. ``Proxy`` + defines a ``__class__`` property, and on PyPy -- with a trace function + installed, i.e. under coverage -- a class body that has such a cell + resolves *every* mention of the name ``__class__`` to the cell rather + than to the class namespace. Reading it then raises ``NameError: + name '__class__' is not defined`` at import time, and binding it + leaves no descriptor on the class at all, so proxies start reporting + themselves instead of the object they wrap. + + Keeping the cell from existing keeps ``__class__`` an ordinary name in + that class body, which is what every interpreter has always handled. + Pinned by tests/unit/test_locals.py::test_Proxy_class_body_bytecode. + """ + super(Proxy, cls).__init_subclass__() + + T = TypeVar("T") S = TypeVar("S") T_co = TypeVar("T_co", covariant=True) @@ -199,7 +221,12 @@ class Proxy(Generic[T]): ) def __init_subclass__(self, source: Optional[type[T]] = None) -> None: - super().__init_subclass__() + # NOTE: Delegated to a module-level helper on purpose -- do not + # inline this back to `super().__init_subclass__()`. Naming `super` + # anywhere in this class body makes the compiler add an implicit + # `__class__` closure cell, which breaks the `__class__` property + # below on PyPy. See `_cooperative_init_subclass`. + _cooperative_init_subclass(self) if source is not None: self._init_from_source(source) elif self.__proxy_source__ is not None: @@ -283,6 +310,9 @@ def __doc__(self) -> Optional[str]: def _get_class(self) -> type[T]: return self._get_current_object().__class__ + # NOTE: This ordinary property spelling is only safe while the class + # body has no implicit `__class__` closure cell -- see + # `_cooperative_init_subclass` before adding any use of `super` here. @property def __class__(self) -> Any: return self._get_class() diff --git a/mode/loop/__init__.py b/mode/loop/__init__.py index 744119d..8020cea 100644 --- a/mode/loop/__init__.py +++ b/mode/loop/__init__.py @@ -24,7 +24,18 @@ mode.loop.use('eventlet') ``` -### gevent +### gevent **deprecated, currently broken** + +!!! warning + This backend is unmaintained, has no test coverage, and does not + presently work on *any* interpreter: selecting it raises + `ImportError: Cannot import 'Loop' from mode.loop._gevent_loop` with + current gevent releases, on GIL-enabled and free-threaded builds + alike. It also re-enables the GIL on free-threaded builds, because + `gevent.libev.corecext` does not declare that it is safe without it. + + Selecting it raises a `DeprecationWarning`, and it will be removed in + a future major release. Use `aio` (the default) or `uvloop`. Use [`gevent`](https://pypi.org/project/gevent) as the event loop. @@ -59,6 +70,7 @@ """ import importlib +import warnings from collections.abc import Mapping from typing import Optional @@ -71,12 +83,34 @@ "uvloop": "mode.loop.uvloop", } +#: Backends that still resolve, but should not be used in new code. +DEPRECATED_LOOPS: Mapping[str, str] = { + "gevent": ( + "The gevent loop backend is deprecated and currently broken: it is " + "unmaintained, has no test coverage, and importing it fails with " + "current gevent releases on every interpreter (see " + "docs/free-threading.md). It also re-enables the GIL on " + "free-threaded builds. Use the 'aio' or 'uvloop' backend instead. " + "It will be removed in a future major release." + ) +} + def use(loop: str) -> None: """Specify the event loop to use as a string. Loop must be one of: aio, eventlet, gevent, uvloop. + + Note: + `gevent` is deprecated and currently broken -- selecting it raises + a `DeprecationWarning` and then fails to import. See the module + docstring. """ + deprecated = DEPRECATED_LOOPS.get(loop) + if deprecated is not None: + # stacklevel=2 attributes this to the caller, so it is actually + # shown when selected from an entrypoint module. + warnings.warn(deprecated, DeprecationWarning, stacklevel=2) mod = LOOPS.get(loop, loop) if mod is not None: importlib.import_module(mod) diff --git a/mode/loop/gevent.py b/mode/loop/gevent.py index 7ac0f22..db8516c 100644 --- a/mode/loop/gevent.py +++ b/mode/loop/gevent.py @@ -1,12 +1,45 @@ -"""Enable [`gevent`](https://pypi.org/project/gevent) support for `asyncio`.""" +"""Enable [`gevent`](https://pypi.org/project/gevent) support for `asyncio`. + +!!! warning "Deprecated and currently broken" + This loop backend is unmaintained, has no test coverage, and does not + presently work on *any* interpreter -- importing it raises + `ImportError: Cannot import 'Loop' from mode.loop._gevent_loop` with + current gevent releases, on GIL-enabled and free-threaded builds + alike. See `docs/free-threading.md` for the diagnosis. + + Use the `aio` (default) or `uvloop` backend instead. This module will + be removed in a future major release. +""" import asyncio import os +import sysconfig import warnings from typing import Optional, cast from mode.utils.loops import get_event_loop +# NOTE: The DeprecationWarning for this backend is raised by +# `mode.loop.use()`, not here. A module-level `warnings.warn` is +# attributed to whichever importlib frame executed the module body, and +# DeprecationWarning is filtered out everywhere except `__main__` -- so it +# would never be shown. Raising it from `use()` with stacklevel=2 puts it +# on the caller instead, which is where users select the backend. + +# NOTE: Deliberately the *build* flag, not `sys._is_gil_enabled()`. The +# runtime check would already read True by the time gevent has been +# imported below, which is exactly the situation being reported. +if sysconfig.get_config_var("Py_GIL_DISABLED"): + warnings.warn( + "The gevent loop is not usable on free-threaded builds: importing " + "gevent re-enables the GIL (gevent.libev.corecext does not declare " + "that it is safe without it), so selecting this loop silently gives " + "up free threading for the whole process. Use the 'aio' or 'uvloop' " + "loop to keep the GIL disabled.", + RuntimeWarning, + stacklevel=2, + ) + os.environ["GEVENT_LOOP"] = "mode.loop._gevent_loop.Loop" try: import gevent diff --git a/mode/signals.py b/mode/signals.py index dad01d6..63c3120 100644 --- a/mode/signals.py +++ b/mode/signals.py @@ -159,7 +159,19 @@ def _get_live_receivers( ) -> tuple[set[SignalHandlerT], set[SignalHandlerRefT]]: live_receivers: set[SignalHandlerT] = set() dead_refs: set[SignalHandlerRefT] = set() - for href in r: + # NOTE: Iterate a snapshot. `r` is the live receiver set shared by + # this signal and every clone of it, and `connect`/`disconnect` + # mutate it from whatever thread or task calls them -- iterating it + # directly raises "Set changed size during iteration". The caller + # also discards dead refs from `r` using what this returns, which + # is itself a mutation during iteration. + # + # It must be `list(r)`, NOT `tuple(r)`: on free-threaded builds + # `list()` (like `set()` and `set.copy()`) takes the source set's + # per-object lock for the duration of the copy, while `tuple()` + # falls back to the generic iterator protocol and does not -- so + # `tuple(r)` raises the very error this snapshot exists to avoid. + for href in list(r): alive, value = self._is_alive(href) if alive and value is not None: live_receivers.add(value) diff --git a/mode/utils/collections.py b/mode/utils/collections.py index 5c5aaf4..a1f2125 100644 --- a/mode/utils/collections.py +++ b/mode/utils/collections.py @@ -2,6 +2,7 @@ import abc import collections.abc +import sys import threading import typing from collections import OrderedDict, UserList @@ -51,6 +52,12 @@ class LazyObject: ... class LazySettings: ... +#: True when running on a free-threaded (:pep:`703`) build with the GIL +#: actually disabled. Checked at runtime rather than build time so that +#: ``PYTHON_GIL=1`` on a free-threaded interpreter is respected. +FREE_THREADED: bool = not getattr(sys, "_is_gil_enabled", lambda: True)() + + __all__ = [ "AttributeDict", "AttributeDictMixin", @@ -438,7 +445,28 @@ class LRUCache(FastUserDict, MutableMapping[KT, VT], MappingViewProxy): the *Least Recently Used* key will be discarded from the cache. thread_safety (bool): Enable if multiple OS threads are going - to access/mutate the cache. + to access/mutate the cache. Defaults to :const:`True` on + free-threaded builds, where there is no GIL to make unguarded + access incidentally safe, and :const:`False` otherwise (which + is what it has always been). It cannot be turned off on a + free-threaded build -- see the note below. + + Note: + The backing store is an :class:`~collections.OrderedDict` rather + than a plain :class:`dict`, even though `dict` has preserved + insertion order since Python 3.7. The reason is + `popitem(last=False)`: evicting the oldest entry is this class's + hot path, and `OrderedDict` does it in O(1) via its linked list, + while the `dict` equivalent (`d.pop(next(iter(d)))`) has to scan + past every slot vacated since the last resize. Measured on a + steady-state evict-and-insert loop, `dict` was ~3x slower at 1,000 + entries and ~110x slower at 100,000. + + The cost of that linked list is that `OrderedDict` is not safe to + mutate concurrently on free-threaded builds -- racing threads + corrupt it badly enough to segfault the interpreter, where `dict` + would merely raise. So on those builds the mutex is mandatory + rather than merely on by default. """ limit: Optional[int] @@ -447,9 +475,26 @@ class LRUCache(FastUserDict, MutableMapping[KT, VT], MappingViewProxy): data: OrderedDict def __init__( - self, limit: Optional[int] = None, *, thread_safety: bool = False + self, + limit: Optional[int] = None, + *, + thread_safety: Optional[bool] = None, ) -> None: self.limit = limit + if thread_safety is None: + thread_safety = FREE_THREADED + elif FREE_THREADED and not thread_safety: + # Not a preference we can honour: an unguarded OrderedDict on a + # free-threaded build is memory-unsafe, not merely racy, and + # taking the interpreter down is a worse outcome than ignoring + # the argument. Say so rather than doing it silently. + raise ValueError( + "LRUCache(thread_safety=False) is not supported on " + "free-threaded builds: the backing OrderedDict can be " + "corrupted by concurrent mutation badly enough to " + "segfault the interpreter. Omit the argument to get the " + "mutex, which is the default here." + ) self.thread_safety = thread_safety self._mutex = self._new_lock() self.data: OrderedDict = OrderedDict() @@ -475,12 +520,31 @@ def popitem(self, *, last: bool = True) -> tuple[KT, VT]: def __setitem__(self, key: KT, value: VT) -> None: # remove least recently used key. with self._mutex: - if self.limit and len(self.data) >= self.limit: + # NOTE: `key not in self.data` matters. Updating a key that is + # already present does not grow the cache, so evicting to make + # room for it discards an unrelated entry for nothing -- a full + # cache would shrink below its own limit on every such update + # (limit=3 holding a/b/c, then `cache["c"] = ...`, used to leave + # two entries and drop "a"). + if ( + key not in self.data + and self.limit + and len(self.data) >= self.limit + ): self.data.pop(next(iter(self.data))) self.data[key] = value + # NOTE: Iteration takes a snapshot under the mutex and yields from that + # snapshot with the mutex released, rather than holding it across the + # yields. Holding a lock across a yield keeps it held for as long as + # the *consumer* takes to iterate -- and forever if the consumer + # abandons the generator half way, since the mutex is only released + # when the generator is closed. Snapshotting also means a concurrent + # writer cannot invalidate an iteration already in progress, which is + # what "dictionary changed size during iteration" used to be. + def __iter__(self) -> Iterator: - return iter(self.data) + return self._keys() def keys(self) -> KeysView[KT]: return ProxyKeysView(self) @@ -488,29 +552,24 @@ def keys(self) -> KeysView[KT]: def _keys(self) -> Iterator[KT]: # userdict.keys in py3k calls __getitem__ with self._mutex: - yield from self.data.keys() + keys = list(self.data) + yield from keys def values(self) -> ValuesView[VT]: return ProxyValuesView(self) def _values(self) -> Iterator[VT]: with self._mutex: - for k in self: - try: - yield self.data[k] - except KeyError: # pragma: no cover - pass + values = list(self.data.values()) + yield from values def items(self) -> ItemsView[KT, VT]: return ProxyItemsView(self) def _items(self) -> Iterator[tuple[KT, VT]]: with self._mutex: - for k in self: - try: - yield (k, self.data[k]) - except KeyError: # pragma: no cover - pass + items = list(self.data.items()) + yield from items def incr(self, key: KT, delta: int = 1) -> int: with self._mutex: diff --git a/mode/utils/objects.py b/mode/utils/objects.py index fb29253..25a2822 100644 --- a/mode/utils/objects.py +++ b/mode/utils/objects.py @@ -3,6 +3,7 @@ import abc import collections.abc import sys +import threading import types import typing from collections.abc import ( @@ -678,6 +679,7 @@ def __init__( self.__name__ = fget.__name__ self.__module__ = fget.__module__ self.class_attribute: Optional[str] = class_attribute + self.__lock = threading.RLock() def is_set(self, obj: Any) -> bool: return self.__name__ in obj.__dict__ @@ -690,8 +692,25 @@ def __get__(self, obj: Any, type: Optional[type] = None) -> RT: try: return cast(RT, obj.__dict__[self.__name__]) except KeyError: - value = obj.__dict__[self.__name__] = self.__get(obj) - return value + pass + # NOTE: The lookup above is the fast path and stays lock-free: once + # the value is cached, reading it is a plain dict hit. Only the + # miss path locks, and it re-checks after acquiring, because + # "look, then compute, then store" is not atomic. Without this, + # two threads that miss together each run `fget` and each store a + # *different* object, so callers disagree about which one is the + # cached one. That is not merely wasted work here: `ServiceProxy` + # documents `@cached_property _service` as the way to build the + # proxied service, and a duplicate there means `start()` and + # `stop()` can act on different Service instances. The GIL made + # this nearly impossible to hit; free-threaded builds hit it + # constantly. + with self.__lock: + try: + return cast(RT, obj.__dict__[self.__name__]) + except KeyError: + value = obj.__dict__[self.__name__] = self.__get(obj) + return value def __set__(self, obj: Any, value: RT) -> None: if self.__set is not None: diff --git a/pyproject.toml b/pyproject.toml index 8de4f1e..0c2a806 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -40,6 +40,7 @@ classifiers = [ "Programming Language :: Python :: 3.12", "Programming Language :: Python :: 3.13", "Programming Language :: Python :: 3.14", + "Programming Language :: Python :: Free Threading :: 2 - Beta", "Programming Language :: Python :: Implementation :: CPython", "Programming Language :: Python :: Implementation :: PyPy", "Operating System :: POSIX", @@ -61,6 +62,12 @@ eventlet = [ "faust-aioeventlet", "dnspython", ] +# DEPRECATED, and currently broken on every interpreter: `mode.loop.use +# ("gevent")` fails with "Cannot import 'Loop' from mode.loop._gevent_loop" +# against current gevent releases. It additionally re-enables the GIL on +# free-threaded (PEP 703) builds, because `gevent.libev.corecext` does not +# declare that it is safe without it. Slated for removal in a future major +# release; use the `aio` or `uvloop` backend. See docs/free-threading.md. gevent = [ "asyncio-gevent~=0.2", ] diff --git a/scripts/tests.sh b/scripts/tests.sh index c8cfba2..7d19dad 100755 --- a/scripts/tests.sh +++ b/scripts/tests.sh @@ -7,4 +7,11 @@ fi set -ex -${PREFIX}pytest tests/unit tests/functional +# Coverage settings (source, omit, branch, fail_under) live in the +# [tool.coverage.*] sections of pyproject.toml. Without --cov nothing is +# measured, which meant the configured `fail_under` was never enforced and +# the Codecov upload in CI had no report to find. +${PREFIX}pytest tests/unit tests/functional \ + --cov \ + --cov-report=term-missing \ + --cov-report=xml diff --git a/tests/freethreading/stress.py b/tests/freethreading/stress.py new file mode 100644 index 0000000..a5960d7 --- /dev/null +++ b/tests/freethreading/stress.py @@ -0,0 +1,428 @@ +"""Free-threading (PEP 703) stress reproducers for mode. + +Every check here should now report ``ok``. Each one reproduced a real +defect before the fix it guards, and they are kept because they are +probabilistic and heavy -- they hammer each surface with 16 threads over +many trials, which is how the ``tuple(r)``-is-not-atomic problem in +`mode.signals` was caught after the first attempt at that fix passed the +cheaper tests. + +The deterministic versions live in +`tests/functional/test_thread_safety.py` and run in CI. This file is +deliberately NOT under the ``testpaths`` configured in ``pyproject.toml``: +before the fixes some of these checks segfaulted the interpreter, and a +regression here should not take the whole test run down with it. + +```sh +uv python install 3.14t +uv venv --python 3.14t .venv-ft +VIRTUAL_ENV=.venv-ft uv pip install -e . -r requirements-tests.txt +.venv-ft/bin/python tests/freethreading/stress.py +``` + +Run it under a GIL-enabled interpreter of the same version too -- the +fixes are meant to hold on both. + +See `docs/free-threading.md` for the measurements and the analysis. +""" + +import sys +import threading +import traceback + +NTHREADS = 16 + + +def race(target, nthreads=NTHREADS): + """Run ``target(i)`` in ``nthreads`` threads released by a barrier. + + Returns the list of tracebacks raised by the threads (empty if none). + """ + barrier = threading.Barrier(nthreads) + errors = [] + + def wrapper(i): + barrier.wait() + try: + target(i) + except BaseException: + errors.append(traceback.format_exc()) + + threads = [ + threading.Thread(target=wrapper, args=(i,)) for i in range(nthreads) + ] + for t in threads: + t.start() + for t in threads: + t.join() + return errors + + +def report(name, errors, note=""): + if errors: + last_line = errors[0].strip().splitlines()[-1] + print(f"[FAIL] {name}: {len(errors)} threads -> {last_line}") + else: + print(f"[ok ] {name} {note}".rstrip()) + return bool(errors) + + +# -------------------------------------------------------------------------- +# Defect 1 (fixed): LRUCache is backed by OrderedDict, whose C linked list +# concurrent mutation can corrupt badly enough to segfault a free-threaded +# interpreter -- and thread_safety defaulted to False. The container is +# unchanged (dict cannot evict the oldest entry in O(1)); instead the mutex +# is now mandatory on free-threaded builds, so the default config is safe +# and thread_safety=False is refused there. +# -------------------------------------------------------------------------- +def check_lru_default(trials=60): + from mode.utils.collections import LRUCache + + print(" (this configuration segfaulted before the fix)", flush=True) + bad = 0 + for _ in range(trials): + cache = LRUCache(limit=50) + + def work(i, cache=cache): + for n in range(100): + cache[f"{i}-{n}"] = n + list(cache.keys()) + + if race(work): + bad += 1 + print( + f"[{'FAIL' if bad else 'ok '}] LRUCache(default): " + f"{bad}/{trials} trials raised" + ) + + +def check_lru_thread_safe(trials=20): + from mode.utils.collections import LRUCache + + bad = 0 + for _ in range(trials): + cache = LRUCache(limit=50, thread_safety=True) + + def work(i, cache=cache): + for n in range(100): + cache[f"{i}-{n}"] = n + list(cache.keys()) + list(cache.items()) + + if race(work): + bad += 1 + print( + f"[{'FAIL' if bad else 'ok '}] LRUCache(thread_safety=True): " + f"{bad}/{trials} trials raised" + ) + + +# -------------------------------------------------------------------------- +# Defect 2 (fixed): cached_property.__get__ was a non-atomic check-then-act +# on obj.__dict__, so racing threads each computed and handed out a distinct +# object. ServiceProxy documents @cached_property as the way to build the +# proxied service, so the duplicate was a real singleton violation. The +# miss path is double-checked under a lock now. +# -------------------------------------------------------------------------- +def check_cached_property(trials=300): + from mode.utils.objects import cached_property + + computes = [0] + bad = 0 + for _ in range(trials): + + class X: + @cached_property + def val(self): + computes[0] += 1 + return object() + + x = X() + seen = [] + lock = threading.Lock() + + def work(i, x=x, seen=seen, lock=lock): + value = x.val + with lock: + seen.append(value) + + race(work) + if len({id(v) for v in seen}) != 1: + bad += 1 + print( + f"[{'FAIL' if bad else 'ok '}] cached_property: {bad}/{trials} " + f"trials returned >1 distinct object " + f"({computes[0]} computes for {trials} properties)" + ) + + +def check_service_proxy(trials=200): + from mode import Service + from mode.proxy import ServiceProxy + from mode.utils.objects import cached_property + + bad = 0 + for _ in range(trials): + built = [] + build_lock = threading.Lock() + + class MyProxy(ServiceProxy): + @cached_property + def _service(self, built=built, build_lock=build_lock): + service = Service() + with build_lock: + built.append(service) + return service + + proxy = MyProxy() + seen = [] + seen_lock = threading.Lock() + + def work(i, proxy=proxy, seen=seen, seen_lock=seen_lock): + service = proxy._service + with seen_lock: + seen.append(service) + + race(work) + if len({id(s) for s in seen}) != 1 or len(built) != 1: + bad += 1 + print( + f"[{'FAIL' if bad else 'ok '}] ServiceProxy._service: " + f"{bad}/{trials} trials built/returned >1 Service instance" + ) + + +# -------------------------------------------------------------------------- +# Defect 4 (fixed): Signal iterated its receiver set while connect/disconnect +# mutated it. Pre-existing -- this failed on GIL builds too. It snapshots +# with list() now (NOT tuple(), which does not lock the source set). +# -------------------------------------------------------------------------- +def check_signal(trials=30): + from mode.signals import Signal + + bad = 0 + for _ in range(trials): + + class Owner: + sig = Signal() + + owner = Owner() + sig = Owner.sig + + def work(i, sig=sig, owner=owner): + for _n in range(100): + + async def handler(*args, **kwargs): + pass + + if i % 2: + sig.connect(handler) + sig.disconnect(handler) + else: + list(sig.iter_receivers(owner)) + + if race(work): + bad += 1 + print( + f"[{'FAIL' if bad else 'ok '}] Signal iter_receivers: " + f"{bad}/{trials} trials raised" + ) + + +# -------------------------------------------------------------------------- +# Surfaces verified SAFE under the same stress -- kept so regressions show up. +# -------------------------------------------------------------------------- +def check_service_subclass_creation(): + from mode import Service + + made = [] + lock = threading.Lock() + + def work(i): + local = [] + for n in range(50): + + async def a_task(self): + pass + + namespace = { + "__module__": f"stressmod{i}", + "__qualname__": f"S{i}_{n}", + "t": Service.task(a_task), + } + local.append(type(f"S{i}_{n}", (Service,), namespace)) + with lock: + made.extend(local) + + errors = race(work) + if not errors: + for cls in made: + clsid = cls._get_class_id() + if cls._tasks.get(clsid) != {"t"}: + errors.append(f"{clsid} -> {cls._tasks.get(clsid)!r}") + report( + "Service subclass creation (cls._tasks)", + errors, + f"({len(made)} classes)", + ) + + +def check_get_event_loop(): + import asyncio + + from mode.utils.loops import get_event_loop + + seen = {} + lock = threading.Lock() + + def work(i): + loops = {get_event_loop() for _ in range(200)} + assert len(loops) == 1, f"thread saw {len(loops)} loops" + with lock: + seen[threading.get_ident()] = loops.pop() + + errors = race(work) + ids = [id(v) for v in seen.values()] + if len(set(ids)) != len(ids): + errors.append("event loop leaked across threads") + for loop in seen.values(): + loop.close() + asyncio.set_event_loop(None) + report("get_event_loop() thread-local cache", errors) + + +def check_service_thread(): + import asyncio + + from mode.threads import ServiceThread + + class T(ServiceThread): + pass + + def work(i): + async def main(): + service = T() + await service.start() + await service.stop() + + asyncio.run(main()) + + report("ServiceThread start/stop", race(work, nthreads=8)) + + +def check_beacon(): + from mode.utils.trees import Node + + root = Node("root") + for i in range(50): + root.new(f"pre-{i}") + + def work(i): + for n in range(200): + if i % 2: + child = root.new(f"{i}-{n}") + root.discard(child.data) + else: + list(root.traverse()) + root.as_graph() + + report("Node.traverse while mutating", race(work)) + + +def check_managed_user_dict(): + from mode.utils.collections import ManagedUserDict + + class D(ManagedUserDict): + def __init__(self): + self.data = {} + + def on_key_get(self, key): ... + def on_key_set(self, key, value): ... + def on_key_del(self, key): ... + def on_clear(self): ... + + d = D() + + def work(i): + for n in range(400): + d[f"{i}-{n}"] = n + d.get(f"{i}-{n}") + del d[f"{i}-{n}"] + + errors = race(work) + if not errors and len(d): + errors.append(f"{len(d)} leftover keys") + report("ManagedUserDict mutation", errors) + + +# -------------------------------------------------------------------------- +# Defect 3 (fixed): mode/__init__.py swapped sys.modules["mode"] for a +# _module instance at the END of its body, so a thread importing mode +# concurrently could be handed the original pre-swap module -- which has no +# __getattr__ -- and every lazily-exported name raised AttributeError. It +# uses a PEP 562 module __getattr__ now, so there is no swap to race with. +# +# Must run in a subprocess: the race only exists on a *cold* import. +# -------------------------------------------------------------------------- +def check_lazy_module(trials=25): + import subprocess + + code = """ +import threading, traceback +errors = [] +barrier = threading.Barrier(16) +names = ["Service", "Worker", "Signal", "Seconds", "get_logger", + "SupervisorStrategy", "label", "want_seconds"] +def work(): + barrier.wait() + try: + import mode + for name in names: + getattr(mode, name) + except BaseException: + errors.append(traceback.format_exc()) +threads = [threading.Thread(target=work) for _ in range(16)] +[t.start() for t in threads] +[t.join() for t in threads] +if errors: + print(errors[0]) + raise SystemExit(1) +""" + bad = 0 + first = "" + for _ in range(trials): + proc = subprocess.run( + [sys.executable, "-c", code], capture_output=True, text=True + ) + if proc.returncode: + bad += 1 + first = first or proc.stdout.strip().splitlines()[-1] + print( + f"[{'FAIL' if bad else 'ok '}] concurrent cold `import mode`: " + f"{bad}/{trials} runs had a failing thread" + + (f" -> {first}" if first else "") + ) + + +def main(): + print(f"python: {sys.version.splitlines()[0]}") + print(f"GIL enabled: {sys._is_gil_enabled()}\n") + + print("-- surfaces verified safe --") + check_service_subclass_creation() + check_get_event_loop() + check_service_thread() + check_beacon() + check_managed_user_dict() + check_lru_thread_safe() + + print("\n-- regression checks (all should be ok) --") + check_lazy_module() + check_signal() + check_cached_property() + check_service_proxy() + check_lru_default() + + +if __name__ == "__main__": + main() diff --git a/tests/functional/test_thread_safety.py b/tests/functional/test_thread_safety.py new file mode 100644 index 0000000..113c65e --- /dev/null +++ b/tests/functional/test_thread_safety.py @@ -0,0 +1,306 @@ +"""Regression tests for the thread-safety fixes. + +These all guard defects that free-threaded (:pep:`703`) builds made +reachable in practice. They are written to fail deterministically on a +GIL-enabled interpreter too, so the whole matrix protects them rather than +just the ``3.14t`` leg. + +See `docs/free-threading.md` for the measurements behind each one, and +`tests/freethreading/stress.py` for the heavier probabilistic reproducers. +""" + +import sys +import threading +import time +from collections import OrderedDict +from types import ModuleType + +import pytest + +import mode +from mode.proxy import ServiceProxy +from mode.signals import Signal +from mode.utils.collections import FREE_THREADED, LRUCache +from mode.utils.objects import cached_property + + +class test_cached_property_is_computed_once: + def _race_on(self, obj, nthreads=8): + barrier = threading.Barrier(nthreads) + seen = [] + lock = threading.Lock() + + def work(): + barrier.wait() + value = obj.val + with lock: + seen.append(value) + + threads = [threading.Thread(target=work) for _ in range(nthreads)] + for t in threads: + t.start() + for t in threads: + t.join() + return seen + + def test_concurrent_miss_computes_once(self): + # The getter sleeps, which releases the GIL, so without the lock in + # `cached_property.__get__` every thread would enter it and store a + # different object. This fails on GIL builds too, by design. + calls = [] + calls_lock = threading.Lock() + + class X: + @cached_property + def val(self): + with calls_lock: + calls.append(1) + time.sleep(0.05) + return object() + + seen = self._race_on(X()) + + assert len(calls) == 1 + assert len({id(v) for v in seen}) == 1 + + def test_service_proxy_service_is_a_singleton(self): + # ServiceProxy documents @cached_property _service as the way to + # build the proxied service, so a duplicate there means start() and + # stop() can act on different Service instances. + built = [] + built_lock = threading.Lock() + + class MyProxy(ServiceProxy): + @cached_property + def _service(self): + service = mode.Service() + with built_lock: + built.append(service) + time.sleep(0.05) + return service + + proxy = MyProxy() + barrier = threading.Barrier(8) + seen = [] + seen_lock = threading.Lock() + + def work(): + barrier.wait() + # Resolve outside the lock -- holding it here would serialise + # the very access this test is trying to race. + service = proxy._service + with seen_lock: + seen.append(service) + + threads = [threading.Thread(target=work) for _ in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert len(built) == 1 + assert len({id(s) for s in seen}) == 1 + + def test_cached_value_is_still_reused(self): + calls = [] + + class X: + @cached_property + def val(self): + calls.append(1) + return object() + + x = X() + assert x.val is x.val + assert len(calls) == 1 + + +class test_LRUCache_thread_safety: + def test_backed_by_ordered_dict(self): + # OrderedDict, not plain dict: evicting the oldest entry is the hot + # path and OrderedDict does it in O(1), where dict has to scan past + # every slot vacated since its last resize. The concurrency + # hazard that comes with it is handled by making the mutex + # mandatory on free-threaded builds, not by changing container. + assert type(LRUCache().data) is OrderedDict + + def test_thread_safety_defaults_to_free_threaded(self): + assert LRUCache().thread_safety is FREE_THREADED + + def test_thread_safety_can_be_requested(self): + assert LRUCache(thread_safety=True).thread_safety is True + + def test_thread_safety_cannot_be_disabled_when_free_threaded(self): + # An unguarded OrderedDict is memory-unsafe here, not merely racy, + # so this is refused rather than honoured. + if FREE_THREADED: + with pytest.raises(ValueError, match="free-threaded"): + LRUCache(thread_safety=False) + else: + assert LRUCache(thread_safety=False).thread_safety is False + + def test_popitem_last_is_lifo(self): + c = LRUCache() + c.update({"a": 1, "b": 2, "c": 3}) + assert c.popitem() == ("c", 3) + assert c.popitem(last=True) == ("b", 2) + + def test_popitem_first_is_fifo(self): + c = LRUCache() + c.update({"a": 1, "b": 2, "c": 3}) + assert c.popitem(last=False) == ("a", 1) + assert c.popitem(last=False) == ("b", 2) + + def test_popitem_empty_raises_KeyError(self): + with pytest.raises(KeyError): + LRUCache().popitem() + with pytest.raises(KeyError): + LRUCache().popitem(last=False) + + def test_limit_still_evicts_oldest(self): + c = LRUCache(limit=3) + for i in range(10): + c[i] = i + assert list(c.keys()) == [7, 8, 9] + + def test_iteration_does_not_hold_the_lock_across_yields(self): + # A half-consumed iterator must not keep the mutex held: the lock + # is reentrant, so only a *different* thread shows the problem. + # Previously the writer below blocked until the abandoned + # generator was collected. + c = LRUCache(limit=100, thread_safety=True) + c.update({"a": 1, "b": 2, "c": 3}) + it = iter(c.keys()) + next(it) # deliberately left half-consumed + + done = threading.Event() + + def writer(): + c["d"] = 4 + done.set() + + thread = threading.Thread(target=writer) + thread.start() + thread.join(timeout=10.0) + + assert done.is_set(), "writer blocked on a half-consumed iterator" + assert c["d"] == 4 + + def test_concurrent_mutation_and_iteration(self): + # Deliberately the *default* configuration -- which on a + # free-threaded build now means the mutex is on. This is the + # workload that used to segfault the interpreter. + c = LRUCache(limit=50) + barrier = threading.Barrier(8) + errors = [] + + def work(i): + barrier.wait() + try: + for n in range(200): + c[f"{i}-{n}"] = n + list(c.keys()) + list(c.items()) + list(c.values()) + except BaseException as exc: # pragma: no cover + errors.append(exc) + + threads = [threading.Thread(target=work, args=(i,)) for i in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert not errors + + +class test_Signal_receiver_iteration: + def test_get_live_receivers_tolerates_mutation(self): + # Directly simulate a connect() landing while the receiver set is + # being walked. Before the snapshot this raised + # "Set changed size during iteration". + signal = Signal() + + async def handler(*args, **kwargs): ... + + for _ in range(4): + signal.connect(handler) + receivers = signal._receivers + original_is_alive = signal._is_alive + + def mutating_is_alive(ref): + receivers.add(lambda: handler) + return original_is_alive(ref) + + signal._is_alive = mutating_is_alive + + live, _dead = signal._get_live_receivers(receivers) + assert live + + def test_iter_receivers_while_connecting(self): + class Owner: + sig = Signal() + + owner = Owner() + signal = Owner.sig + barrier = threading.Barrier(8) + errors = [] + + def work(i): + barrier.wait() + try: + for _n in range(200): + + async def handler(*args, **kwargs): ... + + if i % 2: + signal.connect(handler) + signal.disconnect(handler) + else: + list(signal.iter_receivers(owner)) + except BaseException as exc: # pragma: no cover + errors.append(exc) + + threads = [threading.Thread(target=work, args=(i,)) for i in range(8)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert not errors + + +class test_mode_lazy_imports: + def test_module_is_not_replaced_in_sys_modules(self): + # The old Werkzeug-style trick swapped sys.modules["mode"] for a + # ModuleType *subclass* at the end of mode/__init__.py. That swap + # was the race: a thread importing mode concurrently could be + # handed the original pre-swap module, which had no __getattr__. + # A PEP 562 module __getattr__ needs no swap at all. + assert type(sys.modules["mode"]) is ModuleType + + def test_module_keeps_its_spec(self): + # The replacement module carried no __spec__, which denied the + # import machinery the _initializing flag it uses to make a second + # importing thread wait. + assert mode.__spec__ is not None + assert mode.__spec__.name == "mode" + + def test_lazy_names_resolve(self): + from mode.services import Service + + assert mode.Service is Service + + def test_resolving_one_name_binds_its_siblings(self): + assert mode.task is not None + assert "timer" in vars(mode) + + def test_unknown_attribute_raises_AttributeError(self): + with pytest.raises(AttributeError) as excinfo: + mode.NoSuchThing # noqa: B018 + assert "NoSuchThing" in str(excinfo.value) + + def test_dir_lists_the_lazy_names(self): + listed = dir(mode) + for name in mode.__all__: + assert name in listed diff --git a/tests/functional/utils/test_collections.py b/tests/functional/utils/test_collections.py index 9073bf1..7b66413 100644 --- a/tests/functional/utils/test_collections.py +++ b/tests/functional/utils/test_collections.py @@ -570,6 +570,87 @@ def test_pickle(self, d): assert e == d +class test_LRUCache_ordering: + """Pin the ordering semantics LRUCache depends on. + + Every one of these is a property of the backing mapping rather than of + code in this repo, so they are the assertions most likely to diverge + between interpreters -- CPython, PyPy and free-threaded builds each + implement ordered mappings differently. Keeping them explicit means a + divergence shows up as a named test failure on the relevant leg of the + matrix instead of as mysterious cache behaviour downstream. + """ + + def test_iteration_follows_insertion_order(self): + c = LRUCache() + for key in "abc": + c[key] = key.upper() + assert list(c) == ["a", "b", "c"] + assert list(c.keys()) == ["a", "b", "c"] + assert list(c.values()) == ["A", "B", "C"] + assert list(c.items()) == [("a", "A"), ("b", "B"), ("c", "C")] + + def test_reading_a_key_moves_it_to_the_end(self): + # The LRU touch: this is what makes eviction least-recently-*used* + # rather than merely oldest-inserted. + c = LRUCache() + for key in "abc": + c[key] = key + c["a"] + assert list(c) == ["b", "c", "a"] + + def test_updating_an_existing_key_keeps_its_position(self): + c = LRUCache() + for key in "abc": + c[key] = key + c["a"] = "changed" + assert list(c) == ["a", "b", "c"] + assert c.data["a"] == "changed" + + def test_updating_an_existing_key_does_not_evict(self): + # Regression: __setitem__ used to evict before checking whether the + # key was already present, so updating a key in a full cache + # discarded an unrelated entry and left the cache under its limit. + c = LRUCache(limit=3) + for key in "abc": + c[key] = key + c["c"] = "changed" + assert len(c) == 3 + assert list(c) == ["a", "b", "c"] + + def test_eviction_discards_the_oldest(self): + c = LRUCache(limit=3) + for key in "abcd": + c[key] = key + assert list(c) == ["b", "c", "d"] + + def test_eviction_respects_a_touch(self): + c = LRUCache(limit=3) + for key in "abc": + c[key] = key + c["a"] + c["d"] = "d" + assert list(c) == ["c", "a", "d"] + + def test_update_evicts_the_oldest_first(self): + c = LRUCache(limit=3) + c.update({key: key for key in "abcde"}) + assert list(c) == ["c", "d", "e"] + + def test_popitem_pops_from_either_end(self): + c = LRUCache() + for key in "abc": + c[key] = key + assert c.popitem() == ("c", "c") + assert c.popitem(last=False) == ("a", "a") + + def test_order_survives_a_pickle_round_trip(self): + c = LRUCache() + for key in "abc": + c[key] = key + assert list(pickle.loads(pickle.dumps(c))) == ["a", "b", "c"] + + class test_AttributeDictMixin: @pytest.fixture def d(self): diff --git a/tests/unit/test_locals.py b/tests/unit/test_locals.py index c235880..f600dfb 100644 --- a/tests/unit/test_locals.py +++ b/tests/unit/test_locals.py @@ -1,4 +1,5 @@ import abc +import types from collections.abc import ( AsyncGenerator, AsyncIterable, @@ -11,6 +12,7 @@ Sequence, Set, ) +from pathlib import Path from unittest.mock import MagicMock, Mock import pytest @@ -755,3 +757,66 @@ class ProxySource(Proxy[Source]): s = Source() p = ProxySource(lambda: s) assert p._get_current_object() is s + + +class test_Proxy_class_body_bytecode: + """Guard `Proxy` against the PyPy `__class__` cell bug. + + `Proxy` defines a `__class__` property. That is fine so long as the + class body has no implicit `__class__` closure cell -- but the compiler + adds one as soon as any method in the body so much as *names* `super` + (it cannot tell the zero-argument form from the explicit one). + + With that cell present, PyPy -- and only with a trace function + installed, i.e. under coverage -- resolves every mention of the name + `__class__` in the class body to the cell instead of the class + namespace. Both directions break: + + * reading it (as `@__class__.setter` must) hits the cell while it is + still empty, so importing mode.locals raises + `NameError: name '__class__' is not defined`; + * binding it writes to the cell, so no descriptor is left on the + class and every proxy reports itself instead of the object it + wraps. + + CPython resolves both to the class namespace either way, so it cannot + reproduce any of this -- the compiled class body is the only thing a + CPython-only run can check. `_cooperative_init_subclass` keeps the + cell from being created; these tests keep it that way. + """ + + def _proxy_class_body(self): + import mode.locals + + source = Path(mode.locals.__file__).read_text() + module_code = compile(source, mode.locals.__file__, "exec") + + def walk(code): + for const in code.co_consts: + if isinstance(const, types.CodeType): + yield const + yield from walk(const) + + bodies = [c for c in walk(module_code) if c.co_name == "Proxy"] + assert len(bodies) == 1, "expected exactly one Proxy class body" + return bodies[0] + + def test_class_body_has_no_implicit_class_cell(self): + assert "__class__" not in self._proxy_class_body().co_cellvars, ( + "Proxy's class body has an implicit `__class__` closure cell. " + "Something in it names `super` (or reads `__class__`) inside a " + "method -- even the explicit `super(Proxy, self)` form is " + "enough. That breaks the `__class__` property on PyPy under " + "coverage. Route the call through the module-level " + "`_cooperative_init_subclass` helper instead." + ) + + def test_the_property_is_installed_on_the_class(self): + # The runtime half of the same invariant, and the one that catches + # it on PyPy directly: if `__class__` never lands in the class + # namespace, attribute access falls back to `type.__class__` and + # the proxy reports itself rather than the object it wraps. + assert isinstance(Proxy.__dict__["__class__"], property) + + def test_the_property_still_forwards(self): + assert Proxy(lambda: "hello").__class__ is str diff --git a/tests/unit/test_loop.py b/tests/unit/test_loop.py new file mode 100644 index 0000000..180265f --- /dev/null +++ b/tests/unit/test_loop.py @@ -0,0 +1,72 @@ +import warnings +from contextlib import contextmanager +from unittest.mock import patch + +import pytest + +import mode.loop +from mode.loop import DEPRECATED_LOOPS, LOOPS + + +@contextmanager +def recorded_warnings(): + with warnings.catch_warnings(record=True) as recorded: + warnings.simplefilter("always") + yield recorded + + +class test_use: + # NOTE: `importlib.import_module` is patched out throughout. Actually + # selecting a backend applies process-wide monkey-patches (gevent and + # eventlet both patch the stdlib), which would wreck every test that + # runs afterwards. + + @pytest.mark.parametrize("loop", ["eventlet", "gevent", "uvloop"]) + def test_imports_the_backend_module(self, loop): + with patch("importlib.import_module") as import_module: + with recorded_warnings(): + mode.loop.use(loop) + import_module.assert_called_once_with(LOOPS[loop]) + + def test_aio_imports_nothing(self): + with patch("importlib.import_module") as import_module: + mode.loop.use("aio") + import_module.assert_not_called() + + def test_unknown_name_is_treated_as_a_module_path(self): + with patch("importlib.import_module") as import_module: + mode.loop.use("my.custom.loop") + import_module.assert_called_once_with("my.custom.loop") + + +class test_deprecated_backends: + def test_gevent_is_deprecated(self): + assert "gevent" in DEPRECATED_LOOPS + + def test_use_gevent_warns(self): + with patch("importlib.import_module"): + with pytest.warns( + DeprecationWarning, match="gevent loop backend is deprecated" + ): + mode.loop.use("gevent") + + def test_warning_precedes_the_import(self): + # The backend currently fails to import, so the warning is only of + # any use if it is raised before that happens. + with patch("importlib.import_module", side_effect=ImportError("boom")): + with pytest.warns( + DeprecationWarning, match="gevent loop backend is deprecated" + ): + with pytest.raises(ImportError): + mode.loop.use("gevent") + + @pytest.mark.parametrize("loop", ["aio", "eventlet", "uvloop"]) + def test_other_backends_do_not_warn(self, loop): + with patch("importlib.import_module"): + with recorded_warnings() as recorded: + mode.loop.use(loop) + assert not [ + w + for w in recorded + if issubclass(w.category, DeprecationWarning) + ] diff --git a/tests/unit/utils/test_loops.py b/tests/unit/utils/test_loops.py index 14f661f..4783688 100644 --- a/tests/unit/utils/test_loops.py +++ b/tests/unit/utils/test_loops.py @@ -1,8 +1,20 @@ import asyncio +import contextvars +import signal +import sys import threading -from unittest.mock import patch +from unittest.mock import Mock, patch -from mode.utils.loops import get_event_loop +import pytest + +from mode.utils.loops import ( + _appropriate_signal_handler, + _call_asap, + _is_unix_loop, + call_asap, + clone_loop, + get_event_loop, +) def test_get_event_loop__returns_running_loop_when_running(): @@ -92,3 +104,167 @@ def other_thread() -> None: asyncio.set_event_loop(None) main_loop.close() other_loop_holder["loop"].close() + + +# The helpers below (_is_unix_loop, clone_loop, _appropriate_signal_handler, +# call_asap, _call_asap) had no coverage at all, which left mode/utils/loops.py +# at 34%. They are exported but currently unused inside mode itself. + + +@pytest.fixture +def loop(): + loop = asyncio.new_event_loop() + try: + yield loop + finally: + loop.close() + + +class test_is_unix_loop: + @pytest.mark.skipif( + sys.platform == "win32", reason="no unix event loop on windows" + ) + def test_true_for_a_unix_selector_loop(self, loop): + assert _is_unix_loop(loop) + + def test_false_for_anything_else(self): + assert not _is_unix_loop(Mock(name="loop")) + + +class test_clone_loop: + def test_returns_a_new_loop(self, loop): + new_loop = clone_loop(loop) + try: + assert new_loop is not loop + assert isinstance(new_loop, asyncio.AbstractEventLoop) + finally: + new_loop.close() + + def test_non_unix_loop_copies_no_signal_handlers(self): + new_loop = clone_loop(Mock(name="loop")) + try: + assert isinstance(new_loop, asyncio.AbstractEventLoop) + finally: + new_loop.close() + + @pytest.mark.skipif( + sys.platform == "win32", reason="no signal handlers on windows" + ) + def test_retains_signal_handlers(self, loop): + loop.add_signal_handler(signal.SIGUSR1, lambda: None) + new_loop = clone_loop(loop) + try: + assert signal.SIGUSR1 in new_loop._signal_handlers + finally: + new_loop.remove_signal_handler(signal.SIGUSR1) + new_loop.close() + loop.remove_signal_handler(signal.SIGUSR1) + + +class test_appropriate_signal_handler: + def test_calls_the_original_callback_on_the_parent_loop(self, loop): + called = [] + handle = asyncio.Handle( + lambda *a: called.append(a), + (1, 2), + loop, + contextvars.copy_context(), + ) + + wrapper = _appropriate_signal_handler(loop, handle) + wrapper() + + # _call_asap queues onto the parent loop rather than calling inline. + assert called == [] + assert loop._ready + loop._ready.popleft()._run() + assert called == [(1, 2)] + + +class test_call_asap: + def test_requires_a_loop(self): + with pytest.raises(AssertionError): + call_asap(lambda: None) + + @pytest.mark.skipif( + sys.platform == "win32", reason="no unix event loop on windows" + ) + def test_unix_loop_pushes_to_the_front(self, loop): + # NOTE: Only the ordering is asserted, not the number of calls. + # `_call_asap` currently dispatches the callback twice -- once via + # `loop._call_soon()` and again via the handle it inserts at + # `_ready[0]` -- so "jumped" also shows up at the back. Asserting + # the exact sequence would enshrine that; asserting the front of + # the queue tests the documented contract and keeps passing if the + # duplicate is ever removed. + order = [] + loop.call_soon(lambda: order.append("first-queued")) + call_asap(lambda: order.append("jumped"), loop=loop) + + while loop._ready: + loop._ready.popleft()._run() + + assert order[0] == "jumped" + assert "first-queued" in order + + def test_other_loops_delegate_to_call_soon_threadsafe(self): + mock_loop = Mock(name="loop") + callback = Mock(name="callback") + + result = call_asap(callback, 1, 2, loop=mock_loop) + + mock_loop.call_soon_threadsafe.assert_called_once_with(callback, 1, 2) + assert result is mock_loop.call_soon_threadsafe.return_value + + def test_other_loops_pass_the_context_through(self): + mock_loop = Mock(name="loop") + callback = Mock(name="callback") + context = contextvars.copy_context() + + call_asap(callback, 1, loop=mock_loop, context=context) + + mock_loop.call_soon_threadsafe.assert_called_once_with( + callback, 1, context=context + ) + + +class test__call_asap: + @pytest.mark.skipif( + sys.platform == "win32", reason="no unix event loop on windows" + ) + def test_returns_a_handle_and_wakes_the_loop(self, loop): + callback = Mock(name="callback") + + handle = _call_asap(loop, callback, 1, 2) + + assert isinstance(handle, asyncio.Handle) + assert loop._ready + # Only the front handle is run: `_call_asap` also leaves a second, + # duplicate handle further back in `_ready` (see the note in + # test_unix_loop_pushes_to_the_front). + loop._ready.popleft()._run() + callback.assert_called_once_with(1, 2) + + @pytest.mark.skipif( + sys.platform == "win32", reason="no unix event loop on windows" + ) + def test_accepts_a_context(self, loop): + callback = Mock(name="callback") + + handle = _call_asap(loop, callback, context=contextvars.copy_context()) + + assert isinstance(handle, asyncio.Handle) + + def test_raises_when_the_loop_is_closed(self): + closed = asyncio.new_event_loop() + closed.close() + with pytest.raises(RuntimeError): + _call_asap(closed, Mock(name="callback")) + + @pytest.mark.skipif( + sys.platform == "win32", reason="no unix event loop on windows" + ) + def test_debug_mode_validates_the_callback(self, loop): + loop.set_debug(True) + with pytest.raises(TypeError): + _call_asap(loop, "not-callable")