From 2d7811bc1dc493091676475e4517fef0025aaa3b Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 19:27:36 +0000 Subject: [PATCH 01/12] Support free-threaded CPython (PEP 703) on 3.13t and 3.14t A free-threaded interpreter re-enables the GIL for the whole process the moment it imports an extension module that has not declared `Py_mod_gil = Py_MOD_GIL_NOT_USED`, and reports it only through a RuntimeWarning. All three of faust's Cython extensions were in that state, so importing faust on 3.13t/3.14t silently turned free-threading off: RuntimeWarning: The global interpreter lock (GIL) has been enabled to load module 'faust._cython.windows', which has not declared that it can run safely without the GIL. Set `freethreading_compatible=True` in the three .pyx files, which is what makes Cython emit the slot. The modules qualify: windows.pyx holds cdef doubles written once in __init__ and read-only after, and streams.pyx / conductor.pyx hold per-instance Python references with all shared state in ordinary Python containers. Two things made the loss easy to reintroduce invisibly, so both are pinned down: * The directive only exists in Cython 3.1+, and older Cython ignores unknown directives rather than failing -- a 3.0 build would emit no declaration and no diagnostic. Add a `cython>=3.1` floor for 3.13+ in build-system.requires, and pin cibuildwheel's before-build the same way. * Nothing fails when the declaration is missing. Add tests/unit/test_free_threading.py, which imports each extension in a subprocess and asserts the GIL is still off. The subprocess drops PYTHON_GIL from its environment, or the CI job's PYTHON_GIL=0 would make the assertion vacuous. It skips on a GIL interpreter. With that, the full unit + functional suite passes on both 3.13t and 3.14t against the compiled extensions with the GIL genuinely disabled (2211 passed), so drop the `cp31?t-*` cibuildwheel skip the previous comment said to drop "once faust is verified free-threading-safe", and add `enable = ["cpython-freethreading"]` so cp313t is built alongside cp314t. The new `free-threaded` job covers both interpreters and gates merges, since it is what verifies the wheels being published. Two things it does differently from the other test jobs, both necessary: * It installs requirements/freethreading.txt, not test.txt. Parts of test.txt cannot be built on a free-threaded interpreter at all -- twine pulls in cffi, which refuses to build on 3.13t, and hypothesis 6.130+ ships a PyO3 extension that does not support 3.13t either. The new file documents every omission and pin. * It builds the extensions in place. pytest runs from the repo root, so `import faust` resolves to the source tree, and the accelerated implementations are imported behind `try: ... except ImportError`. Without a .so next to the .pyx the fallback engages silently and the run exercises pure Python regardless of USE_CYTHON -- which is also true of the existing USE_CYTHON=true matrix legs. Document the above in docs/developerguide/free_threading.rst, along with two findings that are not fixed here: aiokafka's extensions have not made the declaration either, so a real worker gets the GIL back when the transport driver loads; and Message.ack/decref is a non-atomic read-modify-write that loses final acks under real parallelism. The latter is not reachable from faust's own code, which acks from the event loop, but is reachable via the public Event.ack. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01K8qT5E3rnSXvw7ibNLXrVr --- .github/workflows/python-package.yml | 65 ++++++++- docs/developerguide/free_threading.rst | 184 +++++++++++++++++++++++++ docs/developerguide/index.rst | 1 + faust/_cython/streams.pyx | 1 + faust/_cython/windows.pyx | 1 + faust/transport/_cython/conductor.pyx | 1 + pyproject.toml | 38 +++-- requirements/freethreading.txt | 53 +++++++ tests/unit/test_free_threading.py | 117 ++++++++++++++++ 9 files changed, 452 insertions(+), 9 deletions(-) create mode 100644 docs/developerguide/free_threading.rst create mode 100644 requirements/freethreading.txt create mode 100644 tests/unit/test_free_threading.py diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml index 79fb3b54a..ea6a43334 100644 --- a/.github/workflows/python-package.yml +++ b/.github/workflows/python-package.yml @@ -138,6 +138,66 @@ jobs: uses: codecov/codecov-action@v5 with: token: ${{ secrets.CODECOV_TOKEN }} + test-freethreading: + name: 'Python ${{ matrix.python-version }} (free-threaded)' + runs-on: ubuntu-latest + timeout-minutes: 15 + strategy: + fail-fast: false + matrix: + # The two interpreters `[tool.cibuildwheel]` publishes free-threaded + # wheels for. Keep the two lists in step: a version we ship a wheel + # for is a version this job has to cover. + python-version: ['3.13t', '3.14t'] + steps: + - uses: actions/checkout@v4 + with: + fetch-depth: 0 + - uses: actions/setup-python@v5 + with: + python-version: ${{ matrix.python-version }} + cache: pip + cache-dependency-path: | + requirements/*.txt + requirements/extras/*.txt + - name: Install dependencies + # Not requirements/test.txt: parts of it cannot be built on a + # free-threaded interpreter at all (twine -> cffi, and hypothesis' + # PyO3 extension on 3.13t). freethreading.txt is that list minus the + # ones that fail, and documents each omission. + run: | + pip install -r requirements/freethreading.txt + pip install 'Cython>=3.1' setuptools setuptools_scm + # Editable, unlike the other jobs' `pip install .`. pytest runs from + # the repo root, so `import faust` resolves to the source tree either + # way -- but the suite also needs the distribution *metadata* to + # exist, because `faust/__init__.py` does + # `version("faust-streaming")` at import time. An editable install + # registers that metadata against the tree the tests actually import, + # instead of a second copy in site-packages that nothing loads. + USE_CYTHON=1 pip install -e . --no-build-isolation + - name: Build the Cython extensions in place + # The extensions have to sit next to the .pyx files or they are never + # imported: `faust/streams.py` and friends pull their accelerated + # implementation in behind `try: ... except ImportError`, so a missing + # .so silently falls back to pure Python and the job would test + # something other than what it thinks. This is also what lets + # tests/unit/test_free_threading.py import the extensions rather than + # skipping. + run: USE_CYTHON=1 python setup.py build_ext --inplace + - name: Verify the extensions did not silently re-enable the GIL + # Fails loudly if an extension is missing `freethreading_compatible`, + # rather than leaving it to a RuntimeWarning nobody reads. Runs + # before the suite so the cause is obvious when it breaks. + run: python -m pytest tests/unit/test_free_threading.py -v --no-cov + - name: Run tests + # PYTHON_GIL=0 keeps the GIL off for the whole run even if some + # *dependency* re-enables it (aiokafka's _crecords does, today), so + # the suite really is exercised without a GIL rather than quietly + # falling back to one. + env: + PYTHON_GIL: '0' + run: python -m pytest tests/unit tests/functional -q --no-cov test-pypy: name: 'Python pypy3.11/Cython: false' runs-on: ubuntu-latest @@ -290,7 +350,10 @@ jobs: check: # This job does nothing and is only used for the branch protection name: ✅ Ensure the required checks passing if: always() - needs: [lint, test-pytest] + # test-freethreading gates too: `[tool.cibuildwheel]` publishes cp313t and + # cp314t wheels, and a wheel we ship should not be able to go out on a red + # run. (The integration jobs stay out of this list -- they are advisory.) + needs: [lint, test-pytest, test-freethreading] runs-on: ubuntu-latest steps: - name: Decide whether the needed jobs succeeded or failed diff --git a/docs/developerguide/free_threading.rst b/docs/developerguide/free_threading.rst new file mode 100644 index 000000000..4fd15255c --- /dev/null +++ b/docs/developerguide/free_threading.rst @@ -0,0 +1,184 @@ +.. _developers-free-threading: + +========================================== + Free-threaded Python (PEP 703) +========================================== + +.. contents:: + :local: + :depth: 2 + +.. _free-threading-status: + +Status +====== + +Faust builds, installs and passes its full unit and functional suite on the +free-threaded builds of CPython 3.13 (``3.13t``) and 3.14 (``3.14t``) with the +GIL genuinely disabled, and ``[tool.cibuildwheel]`` publishes ``cp313t`` and +``cp314t`` wheels. The ``free-threaded`` CI job covers both interpreters and +gates merges. + +What that does **not** mean is that Faust becomes multi-threaded. Faust's +concurrency model is :mod:`asyncio`: agents, streams and the conductor all run +as tasks on a single event loop, and a loop runs one task at a time whether or +not the interpreter has a GIL. Removing the GIL does not make any of that run +in parallel. Treat free-threading support as *"Faust runs correctly on a +free-threaded interpreter"* -- which matters if the rest of your application +wants the no-GIL build -- and not as a throughput feature. + +.. _free-threading-declaration: + +Why the ``.pyx`` files declare ``freethreading_compatible`` +=========================================================== + +A free-threaded interpreter re-enables the GIL, for the whole process, the +moment it imports an extension module that does not declare +``Py_mod_gil = Py_MOD_GIL_NOT_USED``. It reports this with a +:exc:`RuntimeWarning` and nothing else -- the import succeeds, the program +runs, and free-threading is simply gone: + +.. sourcecode:: text + + RuntimeWarning: The global interpreter lock (GIL) has been enabled to load + module 'faust._cython.windows', which has not declared that it can run + safely without the GIL. + +All three extension modules therefore set the directive that makes Cython emit +that slot: + +.. sourcecode:: cython + + # cython: language_level=3 + # cython: freethreading_compatible=True + +Two things make this easy to lose silently, and both are pinned down +deliberately: + +* The directive only exists in **Cython 3.1 and later**. Older Cython + *ignores* unknown directives rather than failing, so building with Cython + 3.0 produces extensions with no declaration and no diagnostic. Hence the + ``cython>=3.1`` floor for Python 3.13+ in ``[build-system].requires`` and the + ``before-build`` pin in ``[tool.cibuildwheel]``. + +* Nothing fails when the declaration is missing. Hence + :file:`tests/unit/test_free_threading.py`, which imports each extension in a + subprocess and asserts the GIL is still off afterwards. It skips entirely on + a normal interpreter, so it costs nothing on the GIL builds. + +The extensions qualify for the declaration because none of them keep mutable +state at C level: :file:`windows.pyx` holds ``cdef`` doubles that are written +once in ``__init__`` and only read afterwards, and :file:`streams.pyx` and +:file:`conductor.pyx` hold per-instance references to Python objects, with all +shared state living in ordinary Python containers that CPython locks +internally. + +.. _free-threading-aiokafka: + +Known limitation: aiokafka re-enables the GIL +============================================= + +Faust's own extensions are clean, but ``aiokafka`` -- a core dependency -- +ships Cython extensions that have not made the declaration. Importing +:mod:`faust` is fine; the GIL comes back when the transport driver is +resolved, which is to say when a worker starts: + +.. sourcecode:: pycon + + >>> import faust, sys + >>> sys._is_gil_enabled() + False + >>> app = faust.App('probe', broker='kafka://localhost:9092') + >>> sys._is_gil_enabled() + False + >>> app.transport # loads aiokafka.record._crecords + >>> sys._is_gil_enabled() + True + +So a real worker on the default ``aiokafka`` transport runs *with* a GIL today, +regardless of anything Faust does. This is an upstream fix, not one Faust can +make. ``PYTHON_GIL=0`` overrides the re-enabling if you want to run without a +GIL anyway -- which is what the CI job does -- but that is an assertion that +``aiokafka``'s extensions are thread-safe, and nobody has verified that. + +.. _free-threading-races: + +Latent races that free-threading would expose +============================================= + +Because everything runs on one event loop, Faust relies in places on +read-modify-write sequences that are not atomic. The event loop serializes +them today, so they are not bugs in normal use, but they *are* the code that +would break first if any of it were ever driven from more than one thread. + +The clearest example is message reference counting, in +:meth:`faust.types.tuples.Message.ack`: + +.. sourcecode:: python + + def ack(self, consumer, n: int = 1) -> bool: + if not self.acked: # check ... + if not self.decref(n): # ... then act + return self.on_final_ack(consumer) + return False + + def decref(self, n: int = 1) -> int: + refcount = self.refcount = max(self.refcount - n, 0) # not atomic + return refcount + +With 16 threads acking the same message on a free-threaded interpreter, the +decrements are lost and the final ack -- the one that marks the offset +safe-to-commit -- either fires more than once or never fires at all. The same +sequence is duplicated in :file:`faust/_cython/streams.pyx` (``after()``), +:file:`faust/streams.py` and :file:`faust/transport/consumer.py`. + +This is reachable from public API: :meth:`faust.Event.ack` is documented for +callers to use, and nothing stops a user calling it from a thread. It is +*not* reachable from Faust's own code paths, all of which ack from the event +loop. Fixing it means either a lock on the ack path -- which is hot, and would +cost every single-threaded user -- or documenting that acking is event-loop-only. +That decision is deliberately left open; it is recorded here so it is not +rediscovered from scratch. + +.. _free-threading-dev-env: + +Working on a free-threaded interpreter +====================================== + +.. sourcecode:: console + + $ uv python install 3.14t + $ uv venv --python 3.14t + $ uv pip install -r requirements/freethreading.txt 'Cython>=3.1' setuptools setuptools_scm + +Use :file:`requirements/freethreading.txt`, not :file:`requirements/test.txt`: +parts of the latter cannot be built on a free-threaded interpreter at all +(``twine`` pulls in ``cffi``, which refuses to build on 3.13t; ``hypothesis`` +6.130+ ships a PyO3 extension that does not support 3.13t either). That file +documents each omission. + +Then build the extensions **in place**: + +.. sourcecode:: console + + $ USE_CYTHON=1 python setup.py build_ext --inplace + $ PYTHON_GIL=0 python -m pytest tests/unit tests/functional + +The in-place build is not optional if you mean to test the compiled code. +:program:`pytest` runs from the repository root, so ``import faust`` resolves to +the source tree, and ``faust/streams.py`` imports its accelerated +implementation behind a ``try: ... except ImportError``: + +.. sourcecode:: python + + if not NO_CYTHON: + try: + from ._cython.streams import StreamIterator as _CStreamIterator + except ImportError: + _CStreamIterator = None + +With no ``.so`` next to the ``.pyx``, that import fails, the fallback engages +silently, and the run exercises the pure-Python path no matter what +``USE_CYTHON`` was set to during ``pip install``. This applies to the +``USE_CYTHON: true`` legs of the main test matrix as well, which is why the +free-threaded job builds in place explicitly. diff --git a/docs/developerguide/index.rst b/docs/developerguide/index.rst index e5d3bbf0f..fbc0df24f 100644 --- a/docs/developerguide/index.rst +++ b/docs/developerguide/index.rst @@ -12,4 +12,5 @@ overview partition_assignor + free_threading diff --git a/faust/_cython/streams.pyx b/faust/_cython/streams.pyx index 42e06bc28..37fc29228 100644 --- a/faust/_cython/streams.pyx +++ b/faust/_cython/streams.pyx @@ -1,4 +1,5 @@ # cython: language_level=3 +# cython: freethreading_compatible=True from asyncio import sleep from time import monotonic diff --git a/faust/_cython/windows.pyx b/faust/_cython/windows.pyx index ea8a66cd2..d591af139 100644 --- a/faust/_cython/windows.pyx +++ b/faust/_cython/windows.pyx @@ -1,4 +1,5 @@ # cython: language_level=3 +# cython: freethreading_compatible=True from datetime import timedelta from libc.math cimport floor from faust.types import WindowT diff --git a/faust/transport/_cython/conductor.pyx b/faust/transport/_cython/conductor.pyx index 9e871e3bf..50f24f83e 100644 --- a/faust/transport/_cython/conductor.pyx +++ b/faust/transport/_cython/conductor.pyx @@ -1,4 +1,5 @@ # cython: language_level=3 +# cython: freethreading_compatible=True from asyncio import ALL_COMPLETED, ensure_future, wait from faust.exceptions import KeyDecodeError, ValueDecodeError diff --git a/pyproject.toml b/pyproject.toml index ee1c27d56..8cddec826 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -68,6 +68,17 @@ requires = [ "wheel", "cython>=0.29; implementation_name == 'cpython'", "cython>=3.0.0; implementation_name == 'cpython' and python_version >= '3.12'", + # 3.1 is the first Cython that understands `freethreading_compatible`, the + # directive the three .pyx files set. It is what makes the generated + # modules emit `Py_mod_gil = Py_MOD_GIL_NOT_USED`; without it a + # free-threaded interpreter re-enables the GIL when it imports them (with + # only a RuntimeWarning to say so), silently undoing free-threading for the + # whole process. Older Cython ignores the directive rather than failing, + # so the floor has to be enforced here or the loss is invisible at build + # time. There is no PEP 508 marker for a free-threaded interpreter, so + # this gates on 3.13 -- the first version with a `t` ABI -- which covers + # every build that could be free-threaded. + "cython>=3.1; implementation_name == 'cpython' and python_version >= '3.13'", ] build-backend = "setuptools.build_meta" @@ -148,14 +159,25 @@ testpaths = [ build = "cp3*" archs = ["auto64"] -# Skip musllinux, and skip the free-threaded builds (cp313t/cp314t): as of -# cibuildwheel 3.1 free-threading is no longer experimental for 3.14, so -# `cp3*` would otherwise build free-threaded wheels for the Cython extension, -# which has not been validated under a no-GIL interpreter. Drop `cp31?t-*` -# from this list once faust is verified free-threading-safe. -skip = ["*musllinux*", "cp31?t-*"] - -before-build = "pip install Cython" +# The free-threaded builds (cp313t/cp314t) used to be skipped here because the +# Cython extension had not been validated under a no-GIL interpreter. It has +# been now: the three .pyx files declare `freethreading_compatible=True`, and +# the full unit + functional suite passes against the compiled extensions on +# both 3.13t and 3.14t with the GIL actually disabled. The `free-threaded` +# CI job keeps that true, and tests/unit/test_free_threading.py fails if an +# extension ever starts re-enabling the GIL again. +# +# Only musllinux stays skipped (unrelated to free-threading). +skip = ["*musllinux*"] + +# cibuildwheel builds cp314t by default but needs to be told to build cp313t. +enable = ["cpython-freethreading"] + +# >=3.1 for the `freethreading_compatible` directive. Older Cython ignores +# unknown directives instead of failing, so a lower version here would quietly +# produce extensions that re-enable the GIL -- see the note in +# `[build-system].requires`. +before-build = "pip install 'Cython>=3.1'" manylinux-x86_64-image = "manylinux2014" manylinux-pypy_x86_64-image = "manylinux2014" diff --git a/requirements/freethreading.txt b/requirements/freethreading.txt new file mode 100644 index 000000000..b72e5141f --- /dev/null +++ b/requirements/freethreading.txt @@ -0,0 +1,53 @@ +# Test dependencies for the free-threaded (PEP 703) CI job. +# +# This is requirements/test.txt with the packages that cannot be installed on a +# free-threaded interpreter removed, and the pins needed to work around them. +# It exists because pip cannot express "install test.txt except for X": every +# omission below is a package that fails to *build*, so it has to be left out +# rather than skipped at runtime. +# +# What is missing compared to test.txt, and why: +# +# * twine. Pulls keyring -> secretstorage -> cryptography -> cffi, and cffi +# refuses to build on free-threaded 3.13 outright ("CFFI does not support +# the free-threaded build of CPython 3.13. Upgrade to free-threaded 3.14 +# or newer"). twine is a publishing tool that the tests never import. +# +# * black / isort / autoflake / flake8* / bandit / pre-commit, and mypy via +# typecheck.txt. Lint-only tools. The `lint` job already runs them on +# PYTHON_LATEST, and running them again here would only re-check the same +# tree with the same pins. Leaving them out also keeps this job away from +# mypy's mypyc-compiled wheels, which are irrelevant to what it tests. +# +# And the one pin that differs: +# +# * hypothesis <6.130 on 3.13t. From 6.130 hypothesis ships a Rust +# extension (hypothesis._native) built with PyO3, and PyO3 does not +# support free-threaded builds before 3.14 ("PyO3 does not support the +# free-threaded build of CPython versions below 3.14"). 3.14t gets a +# working wheel, so the pin is scoped to 3.13. The suite only uses +# hypothesis' pure-Python API, which is unchanged across that boundary. +# +# Everything else in test.txt installs unmodified on both 3.13t and 3.14t. +hypothesis>=3.31; python_version >= '3.14' +hypothesis>=3.31,<6.130; python_version < '3.14' +freezegun>=0.3.11 +pytest-aiofiles>=0.2.0 +pytest-aiohttp>=0.3.0 +pytest-asyncio +pytest-forked +pytest-picked +pytest-cov +pytest-random-order>=0.5.4 +pytest<8 +python-dateutil>=2.8 +pytz>=2018.7 +wheel +intervaltree +-r requirements.txt +-r extras/datadog.txt +-r extras/opentracing.txt +-r extras/redis.txt +-r extras/statsd.txt +-r extras/yaml.txt +-r extras/prometheus.txt diff --git a/tests/unit/test_free_threading.py b/tests/unit/test_free_threading.py new file mode 100644 index 000000000..262fbb9ad --- /dev/null +++ b/tests/unit/test_free_threading.py @@ -0,0 +1,117 @@ +"""Guard the free-threading (PEP 703) properties of the compiled extensions. + +A free-threaded interpreter re-enables the GIL for the whole process the +moment it imports an extension module that does not declare +``Py_mod_gil = Py_MOD_GIL_NOT_USED``. It says so only through a +``RuntimeWarning``, which nothing fails on, so the loss is invisible: the +build succeeds, the tests pass, and free-threading is simply gone at runtime. + +The three ``.pyx`` files set ``# cython: freethreading_compatible=True``, +which is what makes Cython emit that slot. Cython <3.1 does not know the +directive and *ignores* it rather than failing, so a build picking up an +older Cython would drop the declaration silently -- hence the +``cython>=3.1`` floor for 3.13+ in ``pyproject.toml``, and hence this test. + +Everything here is skipped unless the tests are running on a free-threaded +build with the GIL actually disabled, so the module is a no-op on a normal +CPython (and on the ``USE_CYTHON=false`` legs, where there is nothing to +check). +""" + +import sys + +import pytest + +#: The extension modules built from ``faust/**/_cython/*.pyx``. +CYTHON_MODULES = [ + "faust._cython.windows", + "faust._cython.streams", + "faust.transport._cython.conductor", +] + + +def _gil_disabled() -> bool: + # `sys._is_gil_enabled` only exists on 3.13+; on a GIL build it is either + # absent or always True. + is_gil_enabled = getattr(sys, "_is_gil_enabled", None) + return is_gil_enabled is not None and not is_gil_enabled() + + +#: Applied to every test below: there is nothing to assert unless this is a +#: free-threaded interpreter that still has the GIL off by the time the suite +#: runs. Note the GIL can be re-enabled by *any* import that happened earlier +#: (a dependency's extension, for instance), which is exactly the condition +#: this file exists to detect -- but it can only be attributed to faust when +#: faust's own modules are the ones being imported, so the checks below import +#: them in a subprocess. +requires_free_threading = pytest.mark.skipif( + not _gil_disabled(), + reason="not running on a free-threaded interpreter with the GIL disabled", +) + + +def _import_in_subprocess(modules: list) -> "tuple": + """Import `modules` in a fresh interpreter, return (gil_enabled, stderr). + + A subprocess, because the GIL cannot be re-disabled once something has + switched it back on: by the time the test module is imported the damage + from any earlier import is already done and unattributable. + """ + import os + import subprocess + + code = ( + "import sys\n" + + "".join(f"import {m}\n" for m in modules) + + "sys.stdout.write('1' if sys._is_gil_enabled() else '0')\n" + ) + + # Drop PYTHON_GIL from the child's environment. The free-threaded CI job + # runs the suite under `PYTHON_GIL=0`, which forces the GIL to stay off + # even for a module that never declared it was safe -- exactly the thing + # being tested for. Inheriting it would make these assertions pass + # unconditionally, so the child has to see the interpreter's default + # behaviour: re-enable the GIL, and say so on stderr. + env = {k: v for k, v in os.environ.items() if k != "PYTHON_GIL"} + + proc = subprocess.run( + [sys.executable, "-W", "always", "-c", code], + capture_output=True, + text=True, + check=True, + env=env, + ) + return proc.stdout.strip() == "1", proc.stderr + + +@requires_free_threading +@pytest.mark.parametrize("module", CYTHON_MODULES) +def test_extension_does_not_re_enable_gil(module: str) -> None: + """Importing a faust extension must leave the GIL disabled.""" + pytest.importorskip(module, reason="built without Cython (USE_CYTHON=false)") + + gil_enabled, stderr = _import_in_subprocess([module]) + + assert not gil_enabled, ( + f"importing {module} re-enabled the GIL, so the process lost " + f"free-threading. The module is missing the " + f"'# cython: freethreading_compatible=True' directive, or it was " + f"compiled by a Cython older than 3.1 (which ignores that directive). " + f"Interpreter said:\n{stderr}" + ) + + +@requires_free_threading +def test_importing_faust_does_not_re_enable_gil() -> None: + """`import faust` must leave the GIL disabled. + + Broader than the per-module check above: this also catches a *dependency* + imported at faust import time that has not declared itself + free-threading-safe. + """ + gil_enabled, stderr = _import_in_subprocess(["faust"]) + + assert not gil_enabled, ( + f"importing faust re-enabled the GIL. The warning on stderr names " + f"the module responsible:\n{stderr}" + ) From f8fd52050cd2ff4082a89ae0e715d597d449b41a Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 20:43:32 +0000 Subject: [PATCH 02/12] Make the Cython path testable, and fix the bug that hid in it The optional Cython accelerators were never executed by a single test. `pip install .` compiles them into site-packages, but pytest runs from the repository root, so `import faust` resolves to the source tree and every accelerated import sits behind `try: ... except ImportError`. With no .so next to the .pyx the fallback engaged silently, so the `use-cython: true` matrix legs differed from the `false` ones only in whether the build step succeeded. Build the extensions in place on those legs, and add FAUST_REQUIRE_CYTHON, which turns the silent fallback into a failure so the gap cannot quietly reopen. This matters beyond this branch: the parity tests proposed in #751 note they otherwise "just run the pure-Python one twice", which in CI was always. ## The bug this uncovered `StreamIterator._try_get_quick_value` carried two faults that concealed each other. `chan_queue_empty` holds the bound `queue.empty` method: # streams.py # streams.pyx if chan_queue_empty(): if self.chan_queue_empty: A bound method is always truthy, so the extension always reported "queue empty" and took the awaiting path. That made the `else` unreachable -- which hid the fact that it returned the bare value from `get_nowait()` instead of the `(need_slow_get, value)` pair the caller unpacks. Had the fast path ever run, `next()` would have raised TypeError, or silently mis-unpacked a two-element value into `need_slow_get, channel_value`. Both are fixed together; fixing only the condition would have activated the broken return. The pure-Python twin has always had this right, so this restores the fast path the extension was meant to provide and brings the two implementations back into agreement. Net effect: the compiled iterator has been doing strictly more work than the pure Python it was meant to accelerate, for as long as it has existed. ## Tests tests/unit/test_cython_parity.py covers the guard, window parity (HoppingWindow/SlidingWindow against their _Py twins across step boundaries), and both branches of the queue fast path. The stream tests drive `StreamIterator.next()` directly rather than `async for`, which would need a running worker, and count calls to `Channel.__anext__` -- the awaiting path -- because that is the only clean signal. The two obvious alternatives both fail: `get_nowait` is called by `Queue.get` on the slow path too, and `empty` is called from inside `get_nowait`, so both fire either way and only the counts differ. Verified in both directions: reintroducing the bug fails the test with 5 `__anext__` calls for 5 already-queued values, against 0 when fixed. Suite passes in every configuration: extensions built (2254 passed), absent (2207 passed, parity tests skipped), and free-threaded 3.14t with PYTHON_GIL=0 (2258 passed). ## Docs docs/developerguide/cython.rst records how to test the compiled code, the drift history that motivates parity tests (#608, the on_topic_buffer_full defect left unfixed because fixing one twin alone would desynchronise them, and the fast-path pair above), and the conventions for adding an accelerator -- including that the wins concentrate in per-call arithmetic, not in code whose body is mostly `await`. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01K8qT5E3rnSXvw7ibNLXrVr --- .github/workflows/python-package.yml | 22 +++ docs/developerguide/cython.rst | 125 ++++++++++++++ docs/developerguide/free_threading.rst | 8 +- docs/developerguide/index.rst | 1 + faust/_cython/streams.pyx | 21 ++- tests/unit/test_cython_parity.py | 217 +++++++++++++++++++++++++ 6 files changed, 389 insertions(+), 5 deletions(-) create mode 100644 docs/developerguide/cython.rst create mode 100644 tests/unit/test_cython_parity.py diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml index ea6a43334..7d1d02113 100644 --- a/.github/workflows/python-package.yml +++ b/.github/workflows/python-package.yml @@ -125,7 +125,26 @@ jobs: if [ "${{ matrix.kafka-driver }}" = "confluent" ]; then pip install -r requirements/extras/ckafka.txt fi + - name: Build the Cython extensions in place + # `pip install .` above compiles the extensions into site-packages, + # where the tests never see them: pytest runs from the repository + # root, so `import faust` resolves to the source tree, and every + # accelerated import sits behind `try: ... except ImportError`. The + # fallback engaged silently, so these legs differed from the + # `use-cython: false` ones only in whether the build step succeeded -- + # the compiled code itself was never executed by a single test. + # + # Building in place puts the .so files next to the .pyx files, which + # is what the source-tree import actually picks up. + if: matrix.use-cython == 'true' + run: USE_CYTHON=1 python setup.py build_ext --inplace - name: Run tests + # FAUST_REQUIRE_CYTHON turns a silent fallback into a failure, so this + # leg cannot quietly go back to testing pure Python if the build stops + # producing importable extensions. See + # tests/unit/test_cython_parity.py. + env: + FAUST_REQUIRE_CYTHON: ${{ matrix.use-cython == 'true' && '1' || '' }} run: | if [ "${{ matrix.kafka-driver }}" = "confluent" ]; then # Dedicated confluent leg: run just the confluent driver's unit @@ -197,6 +216,9 @@ jobs: # falling back to one. env: PYTHON_GIL: '0' + # As in the main matrix: fail rather than silently fall back to pure + # Python if the extensions stop being importable from the tree. + FAUST_REQUIRE_CYTHON: '1' run: python -m pytest tests/unit tests/functional -q --no-cov test-pypy: name: 'Python pypy3.11/Cython: false' diff --git a/docs/developerguide/cython.rst b/docs/developerguide/cython.rst new file mode 100644 index 000000000..90fac6f83 --- /dev/null +++ b/docs/developerguide/cython.rst @@ -0,0 +1,125 @@ +.. _developers-cython: + +========================================== + The optional Cython accelerators +========================================== + +.. contents:: + :local: + :depth: 2 + +Faust ships several hot code paths twice: a readable pure-Python +implementation, and a Cython one used instead whenever the extension modules +could be built. Nothing in Faust requires the extensions -- every accelerated +import falls back: + +.. sourcecode:: python + + if not NO_CYTHON: + try: + from ._cython.streams import StreamIterator as _CStreamIterator + except ImportError: + _CStreamIterator = None + +That fallback is what makes the accelerators optional, and it is also the +single biggest hazard in maintaining them. This page is about the hazard. + +.. _cython-testing: + +Testing the compiled code +========================= + +**The extensions have to be built in place, or the tests do not touch them.** + +:program:`pytest` runs from the repository root, so ``import faust`` resolves +to the source tree -- not to whatever ``pip install .`` compiled into +``site-packages``. With no ``.so`` next to the ``.pyx``, every accelerated +import raises :exc:`ImportError`, the fallback engages, and the whole suite +tests pure Python. Silently: nothing warns, and the run is green either way. + +.. sourcecode:: console + + $ USE_CYTHON=1 python setup.py build_ext --inplace + $ FAUST_REQUIRE_CYTHON=1 python -m pytest tests/unit tests/functional + +``FAUST_REQUIRE_CYTHON=1`` asserts that the accelerators really were loaded, +turning the silent fallback into a failure. Set it whenever a run is supposed +to be testing the compiled code; the CI legs that build the extensions do. + +Without it, a green run proves nothing about the Cython path, and any test +that compares the two implementations degrades into comparing one +implementation against itself. + +.. _cython-drift: + +Why parity tests exist +====================== + +Two implementations of the same behaviour drift, and this pair has drifted +repeatedly: + +* **#608**, *"Fix cython stream_event_in to match python impl"* -- shipped, and + fixed only after the fact. + +* ``Conductor.on_topic_buffer_full`` passes a channel where a ``TP`` is + expected, so ``Monitor``'s per-TP counts are wrong. The comment in + ``faust/transport/conductor.py`` records that the defect is **deliberately + left unfixed**, because fixing one twin alone would make the two disagree. + The duplication turned a small bug into one nobody wants to touch. + +* ``StreamIterator._try_get_quick_value`` carried two bugs that concealed each + other. ``chan_queue_empty`` holds the bound ``queue.empty`` *method*: + + .. sourcecode:: python + + # streams.py # streams.pyx (before) + if chan_queue_empty(): if self.chan_queue_empty: + + A bound method is always truthy, so the extension always reported "queue + empty" and took the awaiting path. That made the ``else`` branch + unreachable -- which hid the fact that it returned the bare value from + ``get_nowait()`` instead of the ``(need_slow_get, value)`` pair the caller + unpacks. Had the fast path ever run, it would have raised + :exc:`TypeError`, or silently mis-unpacked a two-element value. + + So the extension quietly did *more* work than the pure-Python code it was + meant to accelerate, for as long as it has existed. + +None of these were caught by a test, because until recently no test ever +imported the compiled modules. + +:file:`tests/unit/test_cython_parity.py` covers both halves: it asserts the +accelerators are loaded when they are required, and compares the two +implementations where they can be driven directly. + +.. _cython-writing: + +Writing an accelerator +====================== + +The conventions the existing modules follow: + +* **Keep the pure-Python implementation.** It is the reference, it is what + PyPy and no-compiler installs use, and it is the other half of every parity + test. Name it ``_py_`` or ``_Py`` and export both, so tests can + reach the two independently. + +* **Mirror behaviour rather than approximating it.** Anything the + pure-Python version guarantees -- iteration order, what happens when a + mapping is mutated mid-pass, which exception comes out -- is a guarantee of + the accelerated one too. + +* **Add parity tests in the same change**, parametrised over both + implementations. A differential test over randomised inputs is worth more + than a handful of examples. + +* **Measure first, and record it** -- time the accelerator against its twin in + the same interpreter, and put the numbers in the commit message. An + accelerator that does not clearly pay is a second implementation to keep in + sync forever, in exchange for nothing. (A shared harness for this, + ``extra/tools/benchmark_cython.py``, is proposed in #751.) + +Not every hot path is worth compiling. The wins concentrate in code doing +real per-call arithmetic -- the window types are ~4-5x faster compiled. Code +whose body is mostly ``await`` and calls back into Python gains much less, +because the time is in the awaiting, not the arithmetic. diff --git a/docs/developerguide/free_threading.rst b/docs/developerguide/free_threading.rst index 4fd15255c..4bd625f85 100644 --- a/docs/developerguide/free_threading.rst +++ b/docs/developerguide/free_threading.rst @@ -179,6 +179,8 @@ implementation behind a ``try: ... except ImportError``: With no ``.so`` next to the ``.pyx``, that import fails, the fallback engages silently, and the run exercises the pure-Python path no matter what -``USE_CYTHON`` was set to during ``pip install``. This applies to the -``USE_CYTHON: true`` legs of the main test matrix as well, which is why the -free-threaded job builds in place explicitly. +``USE_CYTHON`` was set to during ``pip install``. + +Set ``FAUST_REQUIRE_CYTHON=1`` to make that a failure rather than a silent +fallback -- every CI leg that builds the extensions does. See +:ref:`developers-cython` for the accelerators in general. diff --git a/docs/developerguide/index.rst b/docs/developerguide/index.rst index fbc0df24f..292bdae02 100644 --- a/docs/developerguide/index.rst +++ b/docs/developerguide/index.rst @@ -12,5 +12,6 @@ overview partition_assignor + cython free_threading diff --git a/faust/_cython/streams.pyx b/faust/_cython/streams.pyx index 37fc29228..c2ccb3f23 100644 --- a/faust/_cython/streams.pyx +++ b/faust/_cython/streams.pyx @@ -189,11 +189,28 @@ cdef class StreamIterator: return None, channel_value, stream_state cdef object _try_get_quick_value(self): + # Returns (need_slow_get, value), matching how ``next()`` unpacks it. + # + # Two bugs used to cancel each other out here. ``chan_queue_empty`` is + # the bound ``queue.empty`` *method*, not a call, so ``if + # self.chan_queue_empty`` tested a method object -- always truthy -- + # and this always reported "queue empty, use the slow path". That made + # the ``else`` unreachable, which hid the fact that it returned the + # bare value from ``get_nowait()`` instead of the (flag, value) pair + # the caller unpacks: had it ever run, ``next()`` would have raised + # TypeError, or silently mis-unpacked a two-element value into + # ``need_slow_get, channel_value``. + # + # Both are fixed together; fixing only the condition would activate the + # broken return. streams.py has always had this right (``if + # chan_queue_empty():`` ... ``channel_value = chan_quick_get()``), so + # this restores the fast path the extension was meant to provide and + # brings the two implementations back into agreement. if self.chan_is_channel: if self.chan_errors: raise self.chan_errors.popleft() - if self.chan_queue_empty: + if self.chan_queue_empty(): return (True, None) else: - return self.chan_quick_get() + return (False, self.chan_quick_get()) return (True, None) diff --git a/tests/unit/test_cython_parity.py b/tests/unit/test_cython_parity.py new file mode 100644 index 000000000..f64cac810 --- /dev/null +++ b/tests/unit/test_cython_parity.py @@ -0,0 +1,217 @@ +"""Check the optional Cython accelerators against their pure-Python twins. + +Faust ships several hot paths twice: a readable pure-Python implementation and +a Cython one used instead whenever the extension could be built (see +``NO_CYTHON``). The two are expected to behave identically, but nothing has +been enforcing that -- and the duplication has already cost real bugs: + +* #608, "Fix cython stream_event_in to match python impl"; +* the ``on_topic_buffer_full`` defect recorded but deliberately left unfixed + in ``faust/transport/conductor.py``, because fixing one twin alone would + make them disagree; +* the ``_try_get_quick_value`` pair fixed alongside this file, where the + extension's queue fast path was both unreachable and, had it run, wrong. + +The last one survived because the compiled code is never imported by the test +suite unless the extensions were built *in place*: pytest runs from the +repository root, so ``import faust`` resolves to the source tree, and every +accelerated import sits behind ``try: ... except ImportError``. A missing +``.so`` therefore means the whole suite silently tests pure Python -- including +the parity tests that exist, which then compare an implementation against +itself. + +``test_cython_is_loaded_when_required`` closes that hole: set +``FAUST_REQUIRE_CYTHON=1`` (the CI legs that build the extensions do) and the +suite fails loudly rather than quietly proving nothing. +""" + +import asyncio +import os + +import pytest + +from faust.windows import ( + HoppingWindow, + SlidingWindow, + _PyHoppingWindow, + _PySlidingWindow, +) + +#: True when faust imported the compiled window type rather than falling back. +CYTHON_LOADED = HoppingWindow is not _PyHoppingWindow + +#: Set by the CI legs that build the extensions. When set, the accelerators +#: are mandatory and their absence is a failure rather than a skip. +REQUIRE_CYTHON = bool(os.environ.get("FAUST_REQUIRE_CYTHON", False)) + +requires_cython = pytest.mark.skipif( + not CYTHON_LOADED, + reason="extensions not built in place (USE_CYTHON=1 python setup.py " + "build_ext --inplace)", +) + + +def test_cython_is_loaded_when_required() -> None: + """Fail when a build that must have the accelerators does not. + + Without this, `USE_CYTHON=true` legs pass just as happily on the + pure-Python fallback, and every parity test below degrades into comparing + an object with itself. + """ + if not REQUIRE_CYTHON: + pytest.skip("FAUST_REQUIRE_CYTHON not set") + assert CYTHON_LOADED, ( + "FAUST_REQUIRE_CYTHON is set, but faust fell back to the pure-Python " + "implementations: the extension modules were not importable from the " + "source tree. Build them with " + "`USE_CYTHON=1 python setup.py build_ext --inplace`; `pip install .` " + "is not enough, because pytest imports faust from the repository root." + ) + + +# --------------------------------------------------------------------- windows +#: (cython, python) pairs. When the extensions are missing both entries are +#: the same object, which is exactly what the guard above exists to catch. +WINDOW_PAIRS = [ + pytest.param(HoppingWindow, _PyHoppingWindow, id="HoppingWindow"), + pytest.param(SlidingWindow, _PySlidingWindow, id="SlidingWindow"), +] + +#: Timestamps chosen to land on, just before and just after step boundaries, +#: where the two implementations' differing arithmetic is most likely to part. +TIMESTAMPS = [0.0, 0.5, 1.0, 4.9, 5.0, 5.1, 9.999, 10.0, 33.3, 100.0, 12345.678] + + +@requires_cython +@pytest.mark.parametrize("cy,py", WINDOW_PAIRS) +@pytest.mark.parametrize("timestamp", TIMESTAMPS) +def test_window_current_matches(cy, py, timestamp) -> None: + if cy is HoppingWindow: + a, b = cy(size=10, step=5, expires=3600), py(size=10, step=5, expires=3600) + else: + a, b = cy(before=10, after=0, expires=3600), py( + before=10, after=0, expires=3600 + ) + assert a.current(timestamp) == pytest.approx(b.current(timestamp)) + + +@requires_cython +@pytest.mark.parametrize("timestamp", TIMESTAMPS) +def test_hopping_window_ranges_matches(timestamp) -> None: + a = HoppingWindow(size=10, step=5, expires=3600) + b = _PyHoppingWindow(size=10, step=5, expires=3600) + assert a.ranges(timestamp) == pytest.approx(b.ranges(timestamp)) + + +@requires_cython +@pytest.mark.parametrize("timestamp", TIMESTAMPS) +def test_hopping_window_stale_matches(timestamp) -> None: + a = HoppingWindow(size=10, step=5, expires=60) + b = _PyHoppingWindow(size=10, step=5, expires=60) + latest = timestamp + 3600 + assert a.stale(timestamp, latest) == b.stale(timestamp, latest) + + +# --------------------------------------------------------------------- streams +def _new_iterator(app): + """A compiled StreamIterator over a fresh channel, plus a call counter. + + Driven directly rather than through ``async for``: ``Stream.__aiter__`` + starts the Stream service and needs a running worker, while + ``StreamIterator.next()`` is exactly the code under test and needs + neither. Plain values (not Events) are used so the assertions stay on + the queue path instead of the flow-control and acking machinery. + + The counter is on the channel's ``__anext__`` -- the awaiting path -- + which gives a clean binary signal: the fast path never touches it. The + two obvious alternatives do not work. ``queue.get_nowait`` is called by + ``Queue.get`` on the slow path too, and ``queue.empty`` is called from + inside ``get_nowait`` as well, so both are called either way and only the + exact counts differ. + """ + from faust.streams import _CStreamIterator + + assert _CStreamIterator is not None, "compiled stream iterator not loaded" + + stream = app.stream(app.channel()) + # `app.stream(channel)` clones the channel, so the queue the iterator + # reads is `stream.channel.queue` -- not the queue of the channel that + # was passed in. + channel = stream.channel + queue = channel.queue + + real_anext = channel.__anext__ + anext_calls = [] + + def counting_anext(): + anext_calls.append(1) + return real_anext() + + channel.__anext__ = counting_anext + # StreamIterator caches the channel's and queue's bound methods at + # construction, so it has to be built after the patch is in place. + return _CStreamIterator(stream), queue, anext_calls + + +@requires_cython +@pytest.mark.asyncio +async def test_cython_stream_uses_queue_fast_path(*, app) -> None: + """The compiled iterator must take the non-blocking queue path. + + ``_try_get_quick_value`` skips the ``await`` when the channel queue + already has something in it. It used to test the truthiness of the bound + ``queue.empty`` method rather than calling it, so the fast path was + unreachable and every value went through ``await __anext__``. + """ + it, queue, anext_calls = _new_iterator(app) + + for i in range(5): + queue.put_nowait(i) + + seen = [] + for _ in range(5): + value, _sensor_state = await asyncio.wait_for(it.next(), timeout=5) + seen.append(value) + + assert seen == [0, 1, 2, 3, 4], ( + "the compiled iterator mis-unpacked the queue fast path: " + "_try_get_quick_value must return (need_slow_get, value)" + ) + assert anext_calls == [], ( + f"the iterator awaited Channel.__anext__ {len(anext_calls)} times for " + f"5 already-queued values: it took the slow path instead of the queue " + f"fast path, so _try_get_quick_value is testing the bound empty " + f"method rather than calling it again" + ) + + +@requires_cython +@pytest.mark.asyncio +async def test_cython_stream_falls_back_to_slow_path_when_empty(*, app) -> None: + """An empty queue must still take the awaiting path. + + The fix to ``_try_get_quick_value`` made the fast path reachable; this is + the other half of the branch, guarding against a fix that always reports + "not empty" and calls ``get_nowait()`` on an empty queue. + + The iterator is only checked to still be *pending* -- completing the slow + path means driving ``Channel.__anext__`` on a channel with no running + worker behind it, which is out of scope here. + """ + it, queue, anext_calls = _new_iterator(app) + assert queue.empty() + + pending = asyncio.ensure_future(it.next()) + try: + # Several turns: next() awaits sleep(0) before it consults the queue. + for _ in range(10): + await asyncio.sleep(0) + assert not pending.done(), ( + "iterator returned a value from an empty queue -- the empty " + "branch of _try_get_quick_value is gone" + ) + assert anext_calls, "iterator did not await Channel.__anext__ on an empty queue" + finally: + pending.cancel() + with pytest.raises(asyncio.CancelledError): + await pending From 1d31f229ed20994583082f6ad8cbd1645fbed4d9 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 21:00:49 +0000 Subject: [PATCH 03/12] Add end-to-end conductor parity tests, and fix the divergence they found The topic conductor is the per-message inner loop of a worker, and it exists twice: `ConductorHandler` in the extension, and the `on_message` closure from `ConductorCompiler.build`. Neither was covered -- the existing conductor tests replace the handler with an AsyncMock and assert it was called, so the fan-out, event reuse, buffer-pressure callbacks, full-queue path and decode-error propagation were untested on both sides. Both handlers take `(conductor, tp, channels)` and are awaited with a Message, so they can be driven over the same input and compared. tests/unit/transport/test_conductor_parity.py does that for each of those paths and diffs a full record of the outcome: which events reached which channels, refcount and acked state, decode counts, and every sensor and consumer callback. Both implementations run against the *same* conductor and the same `channels` set, one after the other, rather than two separately-built environments. `channels` is a set of Topic objects hashed by identity, so two separate sets iterate in unrelated orders, and anything order-sensitive -- which channel decodes first, which ones a mid-fan-out decode error reaches -- would differ for reasons unrelated to the implementations. Sharing the set removes that variable; `reset()` clears queues and recorded callbacks between runs. ## What it found `ConductorHandler` never reused a decoded event. The conductor is supposed to deserialize once and reuse it for every channel whose `(key_type, value_type)` matches, but `event_keyid` was only ever assigned from `_decode()`, which returned it *unchanged* on the first pass. It stayed None forever, the reuse branch was dead, and every subscribed channel re-deserialized the payload. That masked a second fault. Had the keyid ever been set, a mismatched pair fell off the end of `_decode` and returned a bare None, which unpacking into two names raises TypeError on. Fixing the reuse alone would have turned a silent inefficiency into a crash on any topic whose subscribers declare different key or value types -- confirmed by building that partial fix and watching the new heterogeneous-keyid test fail with `TypeError: 'NoneType' object is not iterable` at the unpack. This is the same double-bug shape as `_try_get_quick_value` in streams.pyx, arrived at independently: a dead optimization whose deadness concealed that it was also wrong. It was not only a performance difference. A channel whose event is reused never calls `decode`, so a channel that would have failed to deserialize raised no error under the pure-Python conductor and raised one under the extension -- changing which channels received the message and how many acks it got. The fix ports conductor.py's loop faithfully: `event`/`event_keyid` stay pinned to the first channel, and a channel with a different pair gets its own `dest_event` without displacing the pinned one. `_decode` is gone; `keyid` and `dest_event` were already declared in `__call__` and unused, which suggests this is what it was meant to be. ## Verification 16 parity tests, covering fan-out over 1/2/3 channels, no subscribers, batches, event reuse for matching keyids, per-channel decode for differing keyids, decode errors (whole fan-out and single channel), the full-queue path and the pressure callbacks. Before the fix 6 of them failed, all tracing to that one root cause; after it, all pass. Full suite green in every configuration: extensions built (2270 passed), absent (2207 passed, parity skipped), NO_CYTHON=1, and free-threaded 3.14t under PYTHON_GIL=0 (2274 passed). Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01K8qT5E3rnSXvw7ibNLXrVr --- docs/developerguide/cython.rst | 24 + faust/transport/_cython/conductor.pyx | 32 +- tests/unit/transport/test_conductor_parity.py | 456 ++++++++++++++++++ 3 files changed, 503 insertions(+), 9 deletions(-) create mode 100644 tests/unit/transport/test_conductor_parity.py diff --git a/docs/developerguide/cython.rst b/docs/developerguide/cython.rst index 90fac6f83..0828a3368 100644 --- a/docs/developerguide/cython.rst +++ b/docs/developerguide/cython.rst @@ -85,9 +85,33 @@ repeatedly: So the extension quietly did *more* work than the pure-Python code it was meant to accelerate, for as long as it has existed. +* ``ConductorHandler`` had **the same shape of fault, independently**. The + conductor deserializes a message once and reuses the event for every channel + whose ``(key_type, value_type)`` pair matches. In the extension, + ``event_keyid`` was only ever assigned from ``_decode()``, which returned it + *unchanged* on the first pass -- so it stayed ``None`` forever and the reuse + branch was dead. Every subscribed channel re-deserialized the payload. + + That masked a second fault, again: had the keyid ever been set, a mismatched + pair fell off the end of ``_decode`` and returned a bare ``None``, which + unpacking into two names raises :exc:`TypeError` on. Fixing the reuse alone + would have converted a silent inefficiency into a crash on any topic whose + subscribers declare different key or value types. + + It was not only a performance difference. A channel whose event is reused + never calls ``decode`` at all, so a channel that *would* have failed to + deserialize raised no error under the pure-Python conductor and raised one + under the extension -- changing which channels got the message, and how many + acks the message received. + None of these were caught by a test, because until recently no test ever imported the compiled modules. +The parity suites are :file:`tests/unit/test_cython_parity.py` (windows, the +stream iterator's queue fast path) and +:file:`tests/unit/transport/test_conductor_parity.py` (the conductor's +per-message fan-out, driven end to end through both implementations). + :file:`tests/unit/test_cython_parity.py` covers both halves: it asserts the accelerators are loaded when they are required, and compares the two implementations where they can be driven directly. diff --git a/faust/transport/_cython/conductor.pyx b/faust/transport/_cython/conductor.pyx index 50f24f83e..0e0378b08 100644 --- a/faust/transport/_cython/conductor.pyx +++ b/faust/transport/_cython/conductor.pyx @@ -69,10 +69,31 @@ cdef class ConductorHandler: full = [] try: for chan in channels: - event, event_keyid = self._decode(event, chan, event_keyid) + # Deserialize once and reuse the event for every channel + # whose key/value types match, exactly as conductor.py + # does. `event`/`event_keyid` stay pinned to the first + # channel; a channel with a different type pair gets its + # own event without displacing the pinned one. + # + # This used to go through `_decode()`, which never worked: + # `event_keyid` was only ever assigned from that helper's + # return value, and the helper returned it *unchanged* on + # the first pass, so it stayed None forever and the reuse + # branch was dead -- every channel re-deserialized the + # payload. That in turn masked a second fault: had the + # keyid ever been set, a mismatch fell off the end of + # `_decode` returning a bare None, and unpacking it into + # two names would have raised TypeError. + keyid = (chan.key_type, chan.value_type) if event is None: event = await chan.decode(message, propagate=True) - if not self._put(event, chan, full): + event_keyid = keyid + dest_event = event + elif keyid == event_keyid: + dest_event = event + else: + dest_event = await chan.decode(message, propagate=True) + if not self._put(dest_event, chan, full): continue delivered.add(chan) if full: @@ -109,13 +130,6 @@ cdef class ConductorHandler: def on_pressure_drop(self) -> None: self.consumer_on_buffer_drop(self.tp) - cdef object _decode(self, object event, object channel, object event_keyid): - keyid = channel.key_type, channel.value_type - if event_keyid is None or event is None: - return None, event_keyid - if keyid == event_keyid: - return event, keyid - cdef bint _put(self, object event, object channel, diff --git a/tests/unit/transport/test_conductor_parity.py b/tests/unit/transport/test_conductor_parity.py new file mode 100644 index 000000000..92b3583ff --- /dev/null +++ b/tests/unit/transport/test_conductor_parity.py @@ -0,0 +1,456 @@ +"""End-to-end parity between the two topic-conductor implementations. + +``Conductor._build_handler`` returns one of two objects that are supposed to +behave identically: + +* ``faust.transport._cython.conductor.ConductorHandler`` when the extension + was built, and +* the ``on_message`` closure from ``ConductorCompiler.build`` otherwise. + +Both take ``(conductor, tp, channels)`` and are awaited with a ``Message``, so +they can be driven over the same input and compared -- which is what every test +here does. This is the per-message inner loop of a worker: fan-out to +subscribed channels, event reuse across channels with matching key/value types, +buffer-pressure callbacks, the full-queue path and decode-error propagation. + +The existing conductor tests replace the handler with an ``AsyncMock`` and +assert it was called, so none of that logic was covered on either side. The +duplication has already produced bugs that only differential testing catches -- +see ``docs/developerguide/cython.rst`` -- including one recorded in +``faust/transport/conductor.py`` as deliberately unfixed, because correcting one +twin alone would make the two disagree. + +Both implementations are driven against **the same** conductor and the same +``channels`` set, one after the other, rather than against two separately-built +environments. ``channels`` is a set of ``Topic`` objects hashed by identity, so +two separately-built sets iterate in unrelated orders; anything order-sensitive +(which channel decodes first, which ones a mid-fan-out decode error reaches) +would then differ for reasons that have nothing to do with the implementations. +Sharing the set removes that variable, and ``reset()`` clears the queues and +recorded callbacks between runs. + +Everything here is skipped when the extension is not built, since there is then +only one implementation and a "comparison" would run it against itself. +``FAUST_REQUIRE_CYTHON=1`` turns that skip into a failure. +""" + +import asyncio +from typing import Any, Dict, List, Optional, Set + +import pytest + +from faust.exceptions import KeyDecodeError, ValueDecodeError +from faust.transport.conductor import Conductor, ConductorHandler +from faust.types import TP, Message +from tests.helpers import AsyncMock + +TP1 = TP("foo", 0) + +#: The two implementations, in the order they are run. +IMPLS = ["cython", "python"] + +requires_cython_conductor = pytest.mark.skipif( + ConductorHandler is None, + reason="conductor extension not built in place " + "(USE_CYTHON=1 python setup.py build_ext --inplace)", +) + + +class Harness: + """One conductor and its channels, drivable by either implementation.""" + + def __init__( + self, app: Any, n_channels: int = 1, heterogeneous: bool = False + ) -> None: + self.app = app + + # `app.producer` is a Mock in the unit fixture, so `buffer` is a Mock + # attribute whose `wait_until_ebb()` returns something un-awaitable. + app.producer.buffer.wait_until_ebb = AsyncMock() + # Flow control gates the handler's first await; without this the + # handler blocks forever rather than delivering. + app.flow_control.resume() + + self.conductor = Conductor(app) + # `heterogeneous` gives alternating channels a different `key_type`, so + # their `(key_type, value_type)` pairs differ and the fan-out has to + # deserialize per channel instead of reusing one event. Both types + # decode the same bytes payload, so only the reuse decision changes. + self.channels: List[Any] = [ + app.topic( + f"foo{i}", + value_serializer="raw", + key_type=bytes if (heterogeneous and i % 2) else None, + ) + for i in range(n_channels) + ] + #: The single set both implementations iterate, so ordering matches. + self.channel_set: Set[Any] = set(self.channels) + + # Sensors and consumer callbacks are recorded rather than asserted + # individually, so a difference in *which* callbacks fire shows up as a + # diff instead of being missed. + self.buffer_full_sensor: List[Any] = [] + self.consumer_buffer_full: List[Any] = [] + self.consumer_buffer_drop: List[Any] = [] + self.key_decode_errors: List[Any] = [] + self.value_decode_errors: List[Any] = [] + self.decodes: List[str] = [] + + app.sensors.on_topic_buffer_full = self.buffer_full_sensor.append + app.consumer.on_buffer_full = self.consumer_buffer_full.append + app.consumer.on_buffer_drop = self.consumer_buffer_drop.append + + for chan in self.channels: + chan.on_key_decode_error = self._record(self.key_decode_errors, chan) + chan.on_value_decode_error = self._record(self.value_decode_errors, chan) + self._count_decodes(chan) + + def _record(self, into: List[Any], chan: Any): + async def record(exc: BaseException, message: Message) -> None: + into.append((chan.get_topic_name(), type(exc).__name__)) + + return record + + def _count_decodes(self, chan: Any) -> None: + """Make `decode` calls observable; event reuse is the whole point.""" + real_decode = chan.decode + + async def counting(message, propagate=False): + self.decodes.append(chan.get_topic_name()) + return await real_decode(message, propagate=propagate) + + chan.decode = counting + + def fail_decode(self, exc: BaseException, only: Optional[str] = None) -> None: + """Make decoding raise, for every channel or just one by name.""" + for chan in self.channels: + if only is not None and chan.get_topic_name() != only: + continue + + async def failing(message, propagate=False, *, _c=chan): + self.decodes.append(_c.get_topic_name()) + raise exc + + chan.decode = failing + + def build(self, impl: str, tp: TP = TP1) -> Any: + """The handler under test, built the way the conductor builds it.""" + if impl == "cython": + assert ConductorHandler is not None + return ConductorHandler(self.conductor, tp, self.channel_set) + return self.conductor._compiler.build(self.conductor, tp, self.channel_set) + + def message(self, offset: int = 0, key: bytes = b"k", value: bytes = b"v") -> Any: + return Message( + "foo", + 0, + offset, + 0.0, + 0, + None, + key, + value, + None, + tp=TP1, + generation_id=self.app.consumer_generation_id, + ) + + def reset(self) -> None: + """Clear everything a previous run left behind.""" + for chan in self.channels: + while not chan.queue.empty(): + chan.queue.get_nowait() + for recorded in ( + self.buffer_full_sensor, + self.consumer_buffer_full, + self.consumer_buffer_drop, + self.key_decode_errors, + self.value_decode_errors, + self.decodes, + ): + recorded.clear() + + def drain(self) -> Dict[str, List[Any]]: + """Everything sitting in the channel queues, keyed by topic name.""" + out: Dict[str, List[Any]] = {} + for chan in self.channels: + got = [] + while not chan.queue.empty(): + event = chan.queue.get_nowait() + got.append((event.key, event.value, event.message.offset)) + out[chan.get_topic_name()] = got + return out + + def observations(self, message: Optional[Message] = None) -> Dict[str, Any]: + """The full comparable record of a run.""" + delivered = self.drain() + record: Dict[str, Any] = { + "delivered": delivered, + "n_delivered_total": sum(len(v) for v in delivered.values()), + "n_decodes": len(self.decodes), + "buffer_full_sensor": len(self.buffer_full_sensor), + "consumer_buffer_full": len(self.consumer_buffer_full), + "consumer_buffer_drop": len(self.consumer_buffer_drop), + "key_decode_errors": sorted(self.key_decode_errors), + "value_decode_errors": sorted(self.value_decode_errors), + } + if message is not None: + record["refcount"] = message.refcount + record["acked"] = message.acked + return record + + +@pytest.fixture() +def harness(app, request): + param = getattr(request, "param", 1) + if isinstance(param, tuple): + n, heterogeneous = param + else: + n, heterogeneous = param, False + return Harness(app, n_channels=n, heterogeneous=heterogeneous) + + +def assert_parity(results: Dict[str, Any]) -> None: + cython, python = results["cython"], results["python"] + assert cython == python, ( + f"the Cython conductor and the pure-Python conductor disagree.\n" + f" cython: {cython}\n" + f" python: {python}" + ) + + +async def run_both(harness: Harness, scenario) -> Dict[str, Any]: + """Run `scenario(handler, harness)` under each implementation.""" + results = {} + for impl in IMPLS: + harness.reset() + results[impl] = await scenario(harness.build(impl), harness) + return results + + +# ------------------------------------------------------------------ delivery +@requires_cython_conductor +@pytest.mark.asyncio +@pytest.mark.parametrize("harness", [1, 2, 3], indirect=True) +async def test_parity__fan_out(harness) -> None: + """Every subscribed channel gets the event, and refcount matches.""" + + async def scenario(handler, h): + message = h.message() + await handler(message) + return h.observations(message) + + results = await run_both(harness, scenario) + assert_parity(results) + # ...and the shared expectation, so a pair that agrees but is wrong fails. + n = len(harness.channels) + assert results["cython"]["refcount"] == n + assert results["cython"]["n_delivered_total"] == n + + +@requires_cython_conductor +@pytest.mark.asyncio +async def test_parity__no_channels(harness) -> None: + """A TP with no subscribers must not touch the message.""" + harness.channel_set = set() + + async def scenario(handler, h): + message = h.message() + await handler(message) + return h.observations(message) + + results = await run_both(harness, scenario) + assert_parity(results) + assert results["cython"]["refcount"] == 0 + assert results["cython"]["n_decodes"] == 0 + + +@requires_cython_conductor +@pytest.mark.asyncio +@pytest.mark.parametrize("harness", [3], indirect=True) +async def test_parity__multiple_messages(harness) -> None: + """A batch, to catch state carried between calls.""" + + async def scenario(handler, h): + for offset in range(5): + await handler(h.message(offset=offset, key=f"k{offset}".encode())) + return h.observations() + + results = await run_both(harness, scenario) + assert_parity(results) + assert results["cython"]["n_delivered_total"] == 15 + + +@requires_cython_conductor +@pytest.mark.asyncio +@pytest.mark.parametrize("harness", [2, 4], indirect=True) +async def test_parity__event_reuse_for_matching_keyid(harness) -> None: + """Channels with the same (key_type, value_type) share one decode. + + Both implementations are supposed to deserialize the payload once and reuse + the event for every channel whose key/value types match, so the decode + count is observable behaviour and not an implementation detail: it is the + per-message deserialization cost of a topic with several subscribers. + """ + + async def scenario(handler, h): + message = h.message() + await handler(message) + return h.observations(message) + + results = await run_both(harness, scenario) + assert_parity(results) + # Identical key/value types across all channels: decode once, reuse. + assert results["cython"]["n_decodes"] == 1, ( + f"expected one decode reused across " + f"{len(harness.channels)} same-typed channels, got " + f"{results['cython']['n_decodes']}" + ) + + +@requires_cython_conductor +@pytest.mark.asyncio +@pytest.mark.parametrize("harness", [(2, True), (4, True)], indirect=True) +async def test_parity__no_reuse_for_differing_keyid(harness) -> None: + """Channels with different (key_type, value_type) each decode their own. + + This is the branch the Cython conductor could never reach. `event_keyid` + stayed None, so the mismatch case was dead code -- and it was also wrong: + `_decode` fell off the end returning a bare `None`, which unpacked into two + names would have raised TypeError. Fixing the reuse without this branch + would have turned a silent inefficiency into a crash on any topic whose + subscribers declare different key or value types. + """ + + async def scenario(handler, h): + message = h.message() + await handler(message) + return h.observations(message) + + results = await run_both(harness, scenario) + assert_parity(results) + n = len(harness.channels) + # Half the channels share the pinned event's keyid, half do not: one decode + # for the pinned event plus one per mismatched channel. + assert results["cython"]["n_decodes"] == 1 + n // 2 + assert results["cython"]["n_delivered_total"] == n + + +# -------------------------------------------------------------- decode errors +@requires_cython_conductor +@pytest.mark.asyncio +@pytest.mark.parametrize("harness", [1, 3], indirect=True) +@pytest.mark.parametrize( + "exc_cls,bucket", + [ + (KeyDecodeError, "key_decode_errors"), + (ValueDecodeError, "value_decode_errors"), + ], +) +async def test_parity__decode_error_propagates(harness, exc_cls, bucket) -> None: + """A decode failure must reach every undelivered channel, and ack them. + + This is the branch that acks the message on behalf of the channels that + never received it; getting that count wrong either stalls the commit or + commits past an unprocessed message. + """ + harness.fail_decode(exc_cls("boom")) + + async def scenario(handler, h): + message = h.message() + await handler(message) + return h.observations(message) + + results = await run_both(harness, scenario) + assert_parity(results) + assert len(results["cython"][bucket]) == len(harness.channels) + assert results["cython"]["n_delivered_total"] == 0 + + +@requires_cython_conductor +@pytest.mark.asyncio +@pytest.mark.parametrize("harness", [3], indirect=True) +async def test_parity__decode_error_on_one_channel(harness) -> None: + """One channel's decode fails; the rest of the fan-out must match. + + Which channels were already delivered when the failure lands depends on the + iteration order of the shared `channels` set -- identical for both + implementations here, which is the point of sharing it. + """ + harness.fail_decode(ValueDecodeError("boom"), only="foo1") + + async def scenario(handler, h): + message = h.message() + await handler(message) + return h.observations(message) + + results = await run_both(harness, scenario) + assert_parity(results) + + +# ------------------------------------------------------------ buffer pressure +@requires_cython_conductor +@pytest.mark.asyncio +@pytest.mark.conf(stream_buffer_maxsize=2) +async def test_parity__queue_full_path(harness) -> None: + """When a channel queue is full the handler must await ``chan.put``. + + Filling the queue first drives ``_handle_full``, a separate branch in both + implementations that also fires the ``on_topic_buffer_full`` sensor. + """ + + async def scenario(handler, h): + chan = h.channels[0] + for i in range(2): # stream_buffer_maxsize + chan.queue.put_nowait(f"filler{i}") + assert chan.queue.full() + + message = h.message() + # The put blocks until something drains the queue. + pending = asyncio.ensure_future(handler(message)) + await asyncio.sleep(0) + chan.queue.get_nowait() + chan.queue.get_nowait() + await asyncio.wait_for(pending, timeout=5) + return { + "buffer_full_sensor": len(h.buffer_full_sensor), + "consumer_buffer_full": len(h.consumer_buffer_full), + "consumer_buffer_drop": len(h.consumer_buffer_drop), + "qsize": chan.queue.qsize(), + "refcount": message.refcount, + } + + results = await run_both(harness, scenario) + assert_parity(results) + assert results["cython"][ + "buffer_full_sensor" + ], "the full-queue path did not fire the on_topic_buffer_full sensor" + + +@requires_cython_conductor +@pytest.mark.asyncio +@pytest.mark.conf(stream_buffer_maxsize=8) +async def test_parity__pressure_callbacks(harness) -> None: + """High-pressure and pressure-drop callbacks must fire identically. + + ``put_nowait_enhanced`` invokes ``on_pressure_high`` once the queue passes + its pressure ratio; both implementations pass their own bound callbacks in, + so this checks the wiring rather than the queue. + """ + + async def scenario(handler, h): + chan = h.channels[0] + for offset in range(8): + await handler(h.message(offset=offset)) + # Drain, to drive the pressure-drop callback. + while not chan.queue.empty(): + chan.queue.get_nowait() + return { + "buffer_full_sensor": len(h.buffer_full_sensor), + "consumer_buffer_full": len(h.consumer_buffer_full), + "consumer_buffer_drop": len(h.consumer_buffer_drop), + } + + results = await run_both(harness, scenario) + assert_parity(results) From 31c5e81d2524fee96cf5bc53edc9c2ac2857f750 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 21:31:39 +0000 Subject: [PATCH 04/12] Key topic_buffer_full by TP in both conductors `Monitor.topic_buffer_full` is a `Counter[TP]`, and two paths report into it: the pressure-high callback, which passes a TP, and the full-queue path, which passed the *channel*. The same partition therefore accumulated under two different keys depending on which path noticed the buffer was full -- splitting its count, and adding a second `/stats` entry labelled by channel for a partition already listed by TP. Both implementations had it, which is why it went unfixed for so long: the comment in faust/transport/conductor.py recorded the defect and explicitly declined to fix it, because correcting one twin alone would have made the two disagree. With the parity suite in place that objection is gone -- both are corrected here, together, and the suite holds them level. The `# type: ignore[arg-type]` on the call goes away with it; `mypy -p faust` is clean without it, which is the type checker confirming the argument is now the one the sensor declares. ## Note on what parity testing does not do The conductor parity tests were green throughout, before and after. Both implementations passed the channel, so they agreed with each other perfectly while both were wrong. A differential test only finds *divergence*; a shared mistake is invisible to it. So the coverage added here is deliberately not another comparison: * the full-queue parity test now records the sensor's *argument* rather than a call count, and asserts it equals the TP; * a new test drives a real `Monitor` through the full-queue path and asserts every key of `topic_buffer_full` is a TP. It is parametrised over both implementations rather than comparing them, and runs against the pure-Python conductor even when the extension is absent, since the defect was in both. Verified by reverting both twins and confirming each new assertion fails: `Got: []` and `keyed it by ['Topic']`. Suite green in every configuration: extensions built (2272 passed), absent (2208 passed), free-threaded 3.14t under PYTHON_GIL=0 (2276 passed), and `mypy -p faust` clean. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01K8qT5E3rnSXvw7ibNLXrVr --- docs/developerguide/cython.rst | 24 ++++-- faust/transport/_cython/conductor.pyx | 7 +- faust/transport/conductor.py | 19 ++--- tests/unit/test_cython_parity.py | 6 +- tests/unit/transport/test_conductor_parity.py | 84 ++++++++++++++++--- 5 files changed, 111 insertions(+), 29 deletions(-) diff --git a/docs/developerguide/cython.rst b/docs/developerguide/cython.rst index 0828a3368..4548fa4b0 100644 --- a/docs/developerguide/cython.rst +++ b/docs/developerguide/cython.rst @@ -61,11 +61,25 @@ repeatedly: * **#608**, *"Fix cython stream_event_in to match python impl"* -- shipped, and fixed only after the fact. -* ``Conductor.on_topic_buffer_full`` passes a channel where a ``TP`` is - expected, so ``Monitor``'s per-TP counts are wrong. The comment in - ``faust/transport/conductor.py`` records that the defect is **deliberately - left unfixed**, because fixing one twin alone would make the two disagree. - The duplication turned a small bug into one nobody wants to touch. +* ``Conductor``'s full-queue path passed a channel to + ``on_topic_buffer_full`` where a ``TP`` was expected, so + ``Monitor.topic_buffer_full`` -- a ``Counter[TP]`` -- was keyed by channel + from that path and by ``TP`` from the pressure-high path. The same + partition accumulated under two keys, splitting its count and adding a + second ``/stats`` entry for it. + + Both twins had it, so for a long time the comment in + ``faust/transport/conductor.py`` recorded the defect as **deliberately left + unfixed**: correcting one alone would have made them disagree. The + duplication turned a one-line bug into one nobody wanted to touch. It is + fixed now -- in both, together, which is what the parity suites make safe. + + Worth noting what did *not* catch it: the parity tests were green + throughout, because both implementations were wrong in the same way. A + differential test only finds divergence. Shared mistakes need an assertion + about the behaviour itself, which is why the conductor suite now checks that + the sensor is handed a ``TP`` rather than only that both sides hand it the + same thing. * ``StreamIterator._try_get_quick_value`` carried two bugs that concealed each other. ``chan_queue_empty`` holds the bound ``queue.empty`` *method*: diff --git a/faust/transport/_cython/conductor.pyx b/faust/transport/_cython/conductor.pyx index 0e0378b08..302100c48 100644 --- a/faust/transport/_cython/conductor.pyx +++ b/faust/transport/_cython/conductor.pyx @@ -114,7 +114,12 @@ cdef class ConductorHandler: delivered.add(channel) async def _handle_full(self, event, chan, delivered): - self.on_topic_buffer_full(chan) + # ``self.tp``, not the channel: the sensor takes a ``TP`` (as + # ``on_pressure_high`` below passes), and ``Monitor.topic_buffer_full`` + # is a ``Counter[TP]``. Passing the channel here keyed part of that + # counter by channel instead, so the same partition was counted under + # two different keys depending on which path reported it. + self.on_topic_buffer_full(self.tp) await chan.put(event) delivered.add(chan) diff --git a/faust/transport/conductor.py b/faust/transport/conductor.py index 853e38428..1cc99f83c 100644 --- a/faust/transport/conductor.py +++ b/faust/transport/conductor.py @@ -164,16 +164,15 @@ async def on_message(message: Message) -> None: ) delivered.add(chan) if full: - for _, dest_chan in full: - # XXX wrong argument: ``SensorT.on_topic_buffer_full`` - # takes a ``TP`` (as ``on_pressure_high`` above is - # passed), but a channel is passed here, so - # ``Monitor.topic_buffer_full`` is keyed by channel - # and its per-TP counts are wrong. The Cython twin - # in ``_cython/conductor.pyx`` has the same bug; - # fixing either alone would make them disagree, so - # the defect is only recorded here, not fixed. - on_topic_buffer_full(dest_chan) # type: ignore[arg-type] + for _ in full: + # ``tp``, not the channel: the sensor takes a + # ``TP`` (as ``on_pressure_high`` above passes), + # and ``Monitor.topic_buffer_full`` is a + # ``Counter[TP]``. Passing the channel here keyed + # part of that counter by channel instead, so the + # same partition was counted under two different + # keys depending on which path reported it. + on_topic_buffer_full(tp) await asyncio.wait( [ asyncio.ensure_future(dest_chan.put(dest_event)) diff --git a/tests/unit/test_cython_parity.py b/tests/unit/test_cython_parity.py index f64cac810..c505885c2 100644 --- a/tests/unit/test_cython_parity.py +++ b/tests/unit/test_cython_parity.py @@ -6,9 +6,9 @@ been enforcing that -- and the duplication has already cost real bugs: * #608, "Fix cython stream_event_in to match python impl"; -* the ``on_topic_buffer_full`` defect recorded but deliberately left unfixed - in ``faust/transport/conductor.py``, because fixing one twin alone would - make them disagree; +* the ``on_topic_buffer_full`` defect that sat recorded but unfixed in + ``faust/transport/conductor.py`` for as long as it did precisely because + fixing one twin alone would have made them disagree; * the ``_try_get_quick_value`` pair fixed alongside this file, where the extension's queue fast path was both unreachable and, had it run, wrong. diff --git a/tests/unit/transport/test_conductor_parity.py b/tests/unit/transport/test_conductor_parity.py index 92b3583ff..4f3a8b928 100644 --- a/tests/unit/transport/test_conductor_parity.py +++ b/tests/unit/transport/test_conductor_parity.py @@ -16,9 +16,14 @@ The existing conductor tests replace the handler with an ``AsyncMock`` and assert it was called, so none of that logic was covered on either side. The duplication has already produced bugs that only differential testing catches -- -see ``docs/developerguide/cython.rst`` -- including one recorded in -``faust/transport/conductor.py`` as deliberately unfixed, because correcting one -twin alone would make the two disagree. +see ``docs/developerguide/cython.rst``. + +Note the converse, though: a differential test only finds *divergence*. The +``on_topic_buffer_full`` defect (both implementations passed a channel where the +sensor wanted a ``TP``) kept these comparisons green the whole time it was +present, because both sides were wrong identically. Shared mistakes need an +assertion about the behaviour itself, so a few tests below check what a value +*is* and not only that both implementations produce the same one. Both implementations are driven against **the same** conductor and the same ``channels`` set, one after the other, rather than against two separately-built @@ -40,6 +45,7 @@ import pytest from faust.exceptions import KeyDecodeError, ValueDecodeError +from faust.sensors import Monitor from faust.transport.conductor import Conductor, ConductorHandler from faust.types import TP, Message from tests.helpers import AsyncMock @@ -189,7 +195,9 @@ def observations(self, message: Optional[Message] = None) -> Dict[str, Any]: "delivered": delivered, "n_delivered_total": sum(len(v) for v in delivered.values()), "n_decodes": len(self.decodes), - "buffer_full_sensor": len(self.buffer_full_sensor), + # The arguments, not just the count: what gets passed to + # `on_topic_buffer_full` is the metric's key. + "buffer_full_sensor": list(self.buffer_full_sensor), "consumer_buffer_full": len(self.consumer_buffer_full), "consumer_buffer_drop": len(self.consumer_buffer_drop), "key_decode_errors": sorted(self.key_decode_errors), @@ -398,6 +406,11 @@ async def test_parity__queue_full_path(harness) -> None: Filling the queue first drives ``_handle_full``, a separate branch in both implementations that also fires the ``on_topic_buffer_full`` sensor. + + The sensor argument is checked explicitly, not just for parity. Both + implementations used to pass the *channel* here, so they agreed with each + other and this comparison stayed green while both were wrong -- a shared + mistake is exactly what a differential test cannot see. """ async def scenario(handler, h): @@ -414,18 +427,69 @@ async def scenario(handler, h): chan.queue.get_nowait() await asyncio.wait_for(pending, timeout=5) return { - "buffer_full_sensor": len(h.buffer_full_sensor), - "consumer_buffer_full": len(h.consumer_buffer_full), - "consumer_buffer_drop": len(h.consumer_buffer_drop), + "buffer_full_sensor": list(h.buffer_full_sensor), + "consumer_buffer_full": list(h.consumer_buffer_full), + "consumer_buffer_drop": list(h.consumer_buffer_drop), "qsize": chan.queue.qsize(), "refcount": message.refcount, } results = await run_both(harness, scenario) assert_parity(results) - assert results["cython"][ - "buffer_full_sensor" - ], "the full-queue path did not fire the on_topic_buffer_full sensor" + reported = results["cython"]["buffer_full_sensor"] + assert reported, "the full-queue path did not fire the on_topic_buffer_full sensor" + assert all(arg == TP1 for arg in reported), ( + f"on_topic_buffer_full must be given the TP -- it is the key of " + f"Monitor.topic_buffer_full, a Counter[TP], and the pressure-high path " + f"already passes one. Got: {reported}" + ) + + +@pytest.mark.asyncio +@pytest.mark.conf(stream_buffer_maxsize=2) +@pytest.mark.parametrize("impl", IMPLS) +async def test_monitor_counts_buffer_full_by_tp(app, impl) -> None: + """``Monitor.topic_buffer_full`` must be keyed by TP, from either path. + + The counter is a ``Counter[TP]``, and two code paths report into it: the + pressure-high callback (which always passed a TP) and the full-queue path + (which passed the channel). The same partition therefore accumulated under + two different keys, so per-TP counts were split and ``/stats`` grew a second + entry labelled by channel for the same partition. + + Unlike the parity tests, this asserts the behaviour rather than agreement: + both implementations made the same mistake, so they agreed with each other + throughout. Runs against the pure-Python conductor too, since the defect + was in both. + """ + if impl == "cython" and ConductorHandler is None: + pytest.skip("conductor extension not built in place") + + monitor = Monitor() + app.sensors.add(monitor) + + h = Harness(app, n_channels=1) + # Undo the harness's sensor stub: the real delegate is what is under test. + app.sensors.on_topic_buffer_full = monitor.on_topic_buffer_full + + handler = h.build(impl) + chan = h.channels[0] + for i in range(2): # stream_buffer_maxsize -> forces the full-queue path + chan.queue.put_nowait(f"filler{i}") + + pending = asyncio.ensure_future(handler(h.message())) + await asyncio.sleep(0) + chan.queue.get_nowait() + chan.queue.get_nowait() + await asyncio.wait_for(pending, timeout=5) + + assert monitor.topic_buffer_full, "the full-queue path reported nothing" + bad = [key for key in monitor.topic_buffer_full if not isinstance(key, TP)] + assert not bad, ( + f"Monitor.topic_buffer_full is a Counter[TP], but the {impl} conductor " + f"keyed it by {[type(k).__name__ for k in bad]}: {bad}" + ) + assert monitor.topic_buffer_full[TP1] > 0 @requires_cython_conductor From 03e1b1df007c04caf1c380e9a4f59a58c4c9ba84 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 21:47:23 +0000 Subject: [PATCH 05/12] Put the repaired Cython fast paths behind an opt-in setting Two of the Cython fast paths never ran, each guarded by a condition that could not become true, so the extensions quietly did more work than the Python they were meant to accelerate. Repairing them (in the two PRs below this one) activates code that has by definition never executed in production. Gate it. `cython_optimizations` defaults to False. With it off the extensions behave exactly as the released versions do, so upgrading changes nothing; users opt in per app: app = faust.App('myapp', cython_optimizations=True) or `CYTHON_OPTIMIZATIONS=1` in the environment (prefixed when `env_prefix` is set, like every other env-backed setting). The flag is read once per StreamIterator and once per ConductorHandler -- so once per stream and once per assigned TP, not per message -- into a `bint`, leaving a predictable branch on the hot path rather than an attribute lookup into `app.conf`. ## What it gates, and what it does not Gated: * `StreamIterator._try_get_quick_value` -- taking values already in the channel queue instead of always awaiting. * `ConductorHandler` event reuse -- decoding once and reusing the event across channels with matching key/value types, instead of deserializing once per subscribed channel. Not gated: the `on_topic_buffer_full` argument fix. That one was wrong in *both* implementations, is not Cython-specific, and produced a metric that was simply incorrect -- gating a wrong metric key behind a "Cython improvements" flag would be incoherent. It applies always. ## Consequence worth stating plainly While the setting is off, the Cython and pure-Python paths genuinely differ. That is not new -- it is what has shipped for years -- and the flag does not introduce the divergence, only makes it selectable. The sharpest case is the conductor: a reused event is never decoded again, so a channel whose payload would fail to deserialize raises no error when the event is reused and raises one when it is not, changing which channels receive a message and how many acks it takes. So the parity suites now run with the setting on, which is the configuration in which the two implementations are supposed to agree. Each suite also gains a test pinning the default-off behaviour, so the historical path -- the one most users will actually run -- stays covered: 5 awaits for 5 queued values in the iterator, one decode per channel in the conductor. ## Verification Suite green in every configuration: extensions built (2274 passed), absent (2208 passed), free-threaded 3.14t under PYTHON_GIL=0 (2278 passed). `mypy -p faust` clean, `extra/tools/verify_doc_defaults.py` clean, docs build clean with the setting rendered into the configuration reference. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01K8qT5E3rnSXvw7ibNLXrVr --- docs/developerguide/cython.rst | 44 +++++++++++++++++ docs/includes/settingref.txt | 47 +++++++++++++++++++ faust/_cython/streams.pyx | 11 +++++ faust/transport/_cython/conductor.pyx | 10 +++- faust/types/settings/settings.py | 45 ++++++++++++++++++ tests/unit/test_cython_parity.py | 31 ++++++++++++ tests/unit/transport/test_conductor_parity.py | 43 ++++++++++++++++- 7 files changed, 228 insertions(+), 3 deletions(-) diff --git a/docs/developerguide/cython.rst b/docs/developerguide/cython.rst index 4548fa4b0..844cd3649 100644 --- a/docs/developerguide/cython.rst +++ b/docs/developerguide/cython.rst @@ -24,6 +24,50 @@ import falls back: That fallback is what makes the accelerators optional, and it is also the single biggest hazard in maintaining them. This page is about the hazard. +.. _cython-optin: + +The ``cython_optimizations`` opt-in +=================================== + +Two of the Cython fast paths never ran -- each was guarded by a condition that +could not become true (see :ref:`cython-drift`). Repairing them activates code +that has, by definition, never executed in production, so the repairs are +behind a setting that defaults to **off**: + +.. sourcecode:: python + + app = faust.App('myapp', cython_optimizations=True) + +or ``CYTHON_OPTIMIZATIONS=1`` in the environment (``FAUST_CYTHON_OPTIMIZATIONS`` +when :setting:`env_prefix` is set). With it off, the extensions behave exactly +as the released versions do. + +What it gates: + +* ``StreamIterator._try_get_quick_value`` -- taking values already in the + channel queue instead of always awaiting. +* ``ConductorHandler`` event reuse -- decoding a message once and reusing the + event across channels with matching key/value types, instead of + deserializing once per subscribed channel. + +What it does **not** gate: the ``on_topic_buffer_full`` argument fix. That one +was wrong in *both* implementations, is not a Cython-specific change, and +produces a metric that was simply incorrect before -- so it applies +unconditionally. + +One consequence to be aware of. While the setting is off, the Cython path and +the pure-Python path genuinely differ. That is not new -- it is what has +shipped for years -- and the flag does not introduce the divergence, it makes +it selectable. The sharpest case is in the conductor: a reused event is never +decoded a second time, so a channel whose payload would fail to deserialize +raises no error when the event is reused, and raises one when it is not. That +changes which channels receive a message, and how many acks it takes. + +Consequently the parity suites run with the setting **on** -- that is the +configuration in which the two implementations are supposed to agree. A +separate test in each suite pins the default-off behaviour, so the historical +path stays covered too. + .. _cython-testing: Testing the compiled code diff --git a/docs/includes/settingref.txt b/docs/includes/settingref.txt index 669894cff..84f5a91dd 100644 --- a/docs/includes/settingref.txt +++ b/docs/includes/settingref.txt @@ -282,6 +282,53 @@ the second version is 2, and so on. use: ``app.topic(..., internal=True)``. +.. setting:: cython_optimizations + +``cython_optimizations`` +------------------------ + +:type: :class:`bool` +:default: :const:`False` +:environment: :envvar:`CYTHON_OPTIMIZATIONS` +:version-introduced: 0.12.2 + +Enable the repaired fast paths in the Cython extensions. + +Disabled by default, and has no effect at all unless the optional +Cython extension modules were built. + +Faust ships a few hot paths twice: a pure-Python implementation, and a +Cython one used instead when the extensions are available. Two of the +Cython fast paths never actually ran -- each was guarded by a condition +that could not become true -- so for years the extensions quietly did +more work than the Python they were meant to accelerate: + +* ``StreamIterator`` always awaited the channel rather than taking + values already sitting in the queue, and +* ``ConductorHandler`` re-deserialized the payload once per subscribed + channel instead of decoding once and reusing the event. + +Both are repaired, but the repaired code has by definition never run in +production, so it is opt-in. Leaving this ``False`` keeps the +extensions behaving exactly as the released versions do. + +Note this makes the Cython path differ from the pure-Python path while +disabled -- which has always been true; the flag does not introduce the +divergence, it just makes it selectable. The most visible difference is +in the conductor: a reused event is never decoded again, so a channel +whose payload would fail to deserialize raises no error when the event +is reused, and does when it is not. That changes which channels receive +a message and how many acks it takes. + +Enable it to get the fast paths:: + + app = faust.App('myapp', cython_optimizations=True) + +.. seealso:: + + The developer guide's :ref:`developers-cython` page, for what the + two faults were and how the implementations are held level. + .. setting:: blocking_timeout ``blocking_timeout`` diff --git a/faust/_cython/streams.pyx b/faust/_cython/streams.pyx index c2ccb3f23..b92b90337 100644 --- a/faust/_cython/streams.pyx +++ b/faust/_cython/streams.pyx @@ -35,6 +35,7 @@ cdef class StreamIterator: object topics object acks_enabled_for object _skipped_value + bint cython_optimizations def __init__(self, object stream): self.stream = stream @@ -53,6 +54,9 @@ cdef class StreamIterator: self.unacked = self.consumer._unacked_messages self.add_unacked = self.unacked.add self._skipped_value = self.stream._skipped_value + # Opt-in: see the `cython_optimizations` setting. Read once here + # rather than per message, so the hot path costs a `bint` test. + self.cython_optimizations = bool(self.app.conf.cython_optimizations) if isinstance(self.channel, ChannelT): self.chan_is_channel = True @@ -206,9 +210,16 @@ cdef class StreamIterator: # chan_queue_empty():`` ... ``channel_value = chan_quick_get()``), so # this restores the fast path the extension was meant to provide and # brings the two implementations back into agreement. + # + # Behind the `cython_optimizations` setting, off by default: the + # repaired path has never run in production, so taking it is opt-in. + # Disabled, this reproduces the released behaviour exactly -- always + # reporting "use the slow path", never reaching `get_nowait()`. if self.chan_is_channel: if self.chan_errors: raise self.chan_errors.popleft() + if not self.cython_optimizations: + return (True, None) if self.chan_queue_empty(): return (True, None) else: diff --git a/faust/transport/_cython/conductor.pyx b/faust/transport/_cython/conductor.pyx index 302100c48..3aad2f8de 100644 --- a/faust/transport/_cython/conductor.pyx +++ b/faust/transport/_cython/conductor.pyx @@ -20,6 +20,7 @@ cdef class ConductorHandler: object wait_until_producer_ebb object consumer_on_buffer_full object consumer_on_buffer_drop + bint cython_optimizations def __init__(self, object conductor, object tp, object channels): @@ -33,6 +34,9 @@ cdef class ConductorHandler: self.acquire_flow_control = self.app.flow_control.acquire self.wait_until_producer_ebb = self.app.producer.buffer.wait_until_ebb self.consumer = self.app.consumer + # Opt-in: see the `cython_optimizations` setting. Read once per + # handler (one per assigned TP), not per message. + self.cython_optimizations = bool(self.app.conf.cython_optimizations) # We divide `stream_buffer_maxsize` with Queue.pressure_ratio # find a limit to the number of messages we will buffer # before considering the buffer to be under high pressure. @@ -89,9 +93,13 @@ cdef class ConductorHandler: event = await chan.decode(message, propagate=True) event_keyid = keyid dest_event = event - elif keyid == event_keyid: + elif self.cython_optimizations and keyid == event_keyid: dest_event = event else: + # Reuse is behind the `cython_optimizations` setting, + # off by default: the repaired path has never run in + # production. Disabled, every channel deserializes its + # own event, reproducing the released behaviour. dest_event = await chan.decode(message, propagate=True) if not self._put(dest_event, chan, full): continue diff --git a/faust/types/settings/settings.py b/faust/types/settings/settings.py index 8f68b2cf4..b45892819 100644 --- a/faust/types/settings/settings.py +++ b/faust/types/settings/settings.py @@ -604,6 +604,51 @@ def agent_supervisor(self) -> Type[SupervisorStrategyT]: restarted). """ + @sections.Common.setting( + params.Bool, + version_introduced="0.12.2", + env_name="CYTHON_OPTIMIZATIONS", + default=False, + ) + def cython_optimizations(self) -> bool: + """Enable the repaired fast paths in the Cython extensions. + + Disabled by default, and has no effect at all unless the optional + Cython extension modules were built. + + Faust ships a few hot paths twice: a pure-Python implementation, and a + Cython one used instead when the extensions are available. Two of the + Cython fast paths never actually ran -- each was guarded by a condition + that could not become true -- so for years the extensions quietly did + more work than the Python they were meant to accelerate: + + * ``StreamIterator`` always awaited the channel rather than taking + values already sitting in the queue, and + * ``ConductorHandler`` re-deserialized the payload once per subscribed + channel instead of decoding once and reusing the event. + + Both are repaired, but the repaired code has by definition never run in + production, so it is opt-in. Leaving this ``False`` keeps the + extensions behaving exactly as the released versions do. + + Note this makes the Cython path differ from the pure-Python path while + disabled -- which has always been true; the flag does not introduce the + divergence, it just makes it selectable. The most visible difference is + in the conductor: a reused event is never decoded again, so a channel + whose payload would fail to deserialize raises no error when the event + is reused, and does when it is not. That changes which channels receive + a message and how many acks it takes. + + Enable it to get the fast paths:: + + app = faust.App('myapp', cython_optimizations=True) + + .. seealso:: + + The developer guide's :ref:`developers-cython` page, for what the + two faults were and how the implementations are held level. + """ + @sections.Common.setting( params.Seconds, env_name="BLOCKING_TIMEOUT", diff --git a/tests/unit/test_cython_parity.py b/tests/unit/test_cython_parity.py index c505885c2..86b5d6b39 100644 --- a/tests/unit/test_cython_parity.py +++ b/tests/unit/test_cython_parity.py @@ -155,6 +155,7 @@ def counting_anext(): @requires_cython @pytest.mark.asyncio +@pytest.mark.conf(cython_optimizations=True) async def test_cython_stream_uses_queue_fast_path(*, app) -> None: """The compiled iterator must take the non-blocking queue path. @@ -187,6 +188,7 @@ async def test_cython_stream_uses_queue_fast_path(*, app) -> None: @requires_cython @pytest.mark.asyncio +@pytest.mark.conf(cython_optimizations=True) async def test_cython_stream_falls_back_to_slow_path_when_empty(*, app) -> None: """An empty queue must still take the awaiting path. @@ -215,3 +217,32 @@ async def test_cython_stream_falls_back_to_slow_path_when_empty(*, app) -> None: pending.cancel() with pytest.raises(asyncio.CancelledError): await pending + + +@requires_cython +@pytest.mark.asyncio +async def test_cython_stream_fast_path_is_off_by_default(*, app) -> None: + """Without the opt-in, the iterator behaves as the released versions do. + + `cython_optimizations` defaults to False, so the repaired fast path stays + dormant: every value goes through `await Channel.__anext__` exactly as it + did before the fix. No `conf` marker here on purpose -- this is the + default an unmodified app gets. + """ + assert app.conf.cython_optimizations is False + + it, queue, anext_calls = _new_iterator(app) + for i in range(5): + queue.put_nowait(i) + + seen = [] + for _ in range(5): + value, _sensor_state = await asyncio.wait_for(it.next(), timeout=5) + seen.append(value) + + # Same values either way; only the route differs. + assert seen == [0, 1, 2, 3, 4] + assert len(anext_calls) == 5, ( + f"expected the slow path for all 5 values with the optimizations off, " + f"got {len(anext_calls)} awaits: the fast path is no longer opt-in" + ) diff --git a/tests/unit/transport/test_conductor_parity.py b/tests/unit/transport/test_conductor_parity.py index 4f3a8b928..54c3ef060 100644 --- a/tests/unit/transport/test_conductor_parity.py +++ b/tests/unit/transport/test_conductor_parity.py @@ -240,6 +240,7 @@ async def run_both(harness: Harness, scenario) -> Dict[str, Any]: # ------------------------------------------------------------------ delivery @requires_cython_conductor @pytest.mark.asyncio +@pytest.mark.conf(cython_optimizations=True) @pytest.mark.parametrize("harness", [1, 2, 3], indirect=True) async def test_parity__fan_out(harness) -> None: """Every subscribed channel gets the event, and refcount matches.""" @@ -259,6 +260,7 @@ async def scenario(handler, h): @requires_cython_conductor @pytest.mark.asyncio +@pytest.mark.conf(cython_optimizations=True) async def test_parity__no_channels(harness) -> None: """A TP with no subscribers must not touch the message.""" harness.channel_set = set() @@ -276,6 +278,7 @@ async def scenario(handler, h): @requires_cython_conductor @pytest.mark.asyncio +@pytest.mark.conf(cython_optimizations=True) @pytest.mark.parametrize("harness", [3], indirect=True) async def test_parity__multiple_messages(harness) -> None: """A batch, to catch state carried between calls.""" @@ -292,6 +295,7 @@ async def scenario(handler, h): @requires_cython_conductor @pytest.mark.asyncio +@pytest.mark.conf(cython_optimizations=True) @pytest.mark.parametrize("harness", [2, 4], indirect=True) async def test_parity__event_reuse_for_matching_keyid(harness) -> None: """Channels with the same (key_type, value_type) share one decode. @@ -319,6 +323,7 @@ async def scenario(handler, h): @requires_cython_conductor @pytest.mark.asyncio +@pytest.mark.conf(cython_optimizations=True) @pytest.mark.parametrize("harness", [(2, True), (4, True)], indirect=True) async def test_parity__no_reuse_for_differing_keyid(harness) -> None: """Channels with different (key_type, value_type) each decode their own. @@ -348,6 +353,7 @@ async def scenario(handler, h): # -------------------------------------------------------------- decode errors @requires_cython_conductor @pytest.mark.asyncio +@pytest.mark.conf(cython_optimizations=True) @pytest.mark.parametrize("harness", [1, 3], indirect=True) @pytest.mark.parametrize( "exc_cls,bucket", @@ -378,6 +384,7 @@ async def scenario(handler, h): @requires_cython_conductor @pytest.mark.asyncio +@pytest.mark.conf(cython_optimizations=True) @pytest.mark.parametrize("harness", [3], indirect=True) async def test_parity__decode_error_on_one_channel(harness) -> None: """One channel's decode fails; the rest of the fan-out must match. @@ -400,7 +407,7 @@ async def scenario(handler, h): # ------------------------------------------------------------ buffer pressure @requires_cython_conductor @pytest.mark.asyncio -@pytest.mark.conf(stream_buffer_maxsize=2) +@pytest.mark.conf(cython_optimizations=True, stream_buffer_maxsize=2) async def test_parity__queue_full_path(harness) -> None: """When a channel queue is full the handler must await ``chan.put``. @@ -494,7 +501,7 @@ async def test_monitor_counts_buffer_full_by_tp(app, impl) -> None: @requires_cython_conductor @pytest.mark.asyncio -@pytest.mark.conf(stream_buffer_maxsize=8) +@pytest.mark.conf(cython_optimizations=True, stream_buffer_maxsize=8) async def test_parity__pressure_callbacks(harness) -> None: """High-pressure and pressure-drop callbacks must fire identically. @@ -518,3 +525,35 @@ async def scenario(handler, h): results = await run_both(harness, scenario) assert_parity(results) + + +@requires_cython_conductor +@pytest.mark.asyncio +@pytest.mark.parametrize("harness", [3], indirect=True) +async def test_event_reuse_is_off_by_default(harness) -> None: + """Without the opt-in, the conductor behaves as the released versions do. + + `cython_optimizations` defaults to False, so the repaired reuse stays + dormant and every channel deserializes its own event -- exactly as before + the fix. No `conf` marker here on purpose: this is what an unmodified app + gets. + + This is also where the Cython and pure-Python conductors legitimately + differ, so it is not a parity test. That divergence is not new; the flag + only makes it selectable. + """ + assert harness.app.conf.cython_optimizations is False + + handler = harness.build("cython") + message = harness.message() + await handler(message) + obs = harness.observations(message) + + n = len(harness.channels) + assert obs["n_decodes"] == n, ( + f"expected one decode per channel with the optimizations off, got " + f"{obs['n_decodes']} for {n} channels: reuse is no longer opt-in" + ) + # Delivery itself is unchanged -- only how many times the payload is read. + assert obs["n_delivered_total"] == n + assert obs["refcount"] == n From 811d81c1dc1a84be7d31b23ef74a0c610067ac61 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 22:01:08 +0000 Subject: [PATCH 06/12] Make retiring cython_optimizations a two-line change The setting is transitional -- it exists so the repaired Cython fast paths are adopted deliberately rather than arriving in an upgrade, and it is meant to be removed, not kept. Retiring it naively has a trap in it, which this closes before anyone walks into it. `Param.__get__` emits a UserWarning on *every read* of a setting once `version_deprecated` is set, and faust reads this one itself: once per Stream, once per assigned partition. Setting `version_deprecated` would therefore make faust warn at itself, at a rate that scales with the deployment, about a setting the user most likely never set and cannot act on. Measured before the change: three StreamIterator constructions, three warnings. Both extensions now read the flag through `faust.utils.optin.cython_optimizations_enabled`, which takes the value the descriptor stores instead of going through the descriptor. Internal reads stay silent; `app.conf.cython_optimizations` still warns, which is the entire point of deprecating a setting -- a helper that disarmed that too would be worse than the noise, because nobody would ever be told to stop using it. The storage attribute is looked up through the settings registry rather than hard-coded, so renaming the setting cannot silently turn this into a read of a missing attribute. Deliberately not `warnings.catch_warnings()`: it manipulates global state and is not thread-safe, which matters on the free-threaded builds this branch series added support for. ## Tests tests/unit/utils/test_optin.py pins both halves of the contract -- the internal read silent under deprecation, the public read still warning -- plus an end-to-end check that three stream iterators and three conductor handlers produce zero warnings with the setting marked deprecated (three and three before). The deprecation is applied by a fixture that restores the param afterwards, so the tests need no released deprecation to run. ## Docs The developer guide gains the intended sequence: ships off, default flipped once there is real-world evidence (parity passing is necessary but not sufficient -- it only proves the two implementations agree under test), deprecated, then removed along with the branches, the helper and the default-off tests. The setting's own docstring says it is transitional, so it does not read as permanent API. Suite green in every configuration: extensions built (2280 passed), absent (2213 passed), free-threaded 3.14t under PYTHON_GIL=0 (2284 passed). mypy, verify_doc_defaults and the docs build all clean. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01K8qT5E3rnSXvw7ibNLXrVr --- docs/developerguide/cython.rst | 37 +++++++++ docs/includes/settingref.txt | 8 ++ faust/_cython/streams.pyx | 5 +- faust/transport/_cython/conductor.pyx | 7 +- faust/types/settings/settings.py | 8 ++ faust/utils/optin.py | 48 +++++++++++ tests/unit/utils/test_optin.py | 115 ++++++++++++++++++++++++++ 7 files changed, 225 insertions(+), 3 deletions(-) create mode 100644 faust/utils/optin.py create mode 100644 tests/unit/utils/test_optin.py diff --git a/docs/developerguide/cython.rst b/docs/developerguide/cython.rst index 844cd3649..7facfde9f 100644 --- a/docs/developerguide/cython.rst +++ b/docs/developerguide/cython.rst @@ -68,6 +68,43 @@ configuration in which the two implementations are supposed to agree. A separate test in each suite pins the default-off behaviour, so the historical path stays covered too. +.. _cython-optin-lifecycle: + +Retiring the setting +-------------------- + +The setting is **transitional**. It exists to make adopting the repaired +paths a decision rather than something that arrives in an upgrade, and it is +meant to be removed, not kept. The intended sequence: + +1. **Now** -- ships defaulting to ``False``. Upgrading changes nothing. +2. **Default flipped to** ``True`` once there is real-world evidence the + repaired paths behave: the parity suites passing is necessary but not + sufficient, since they only prove the two implementations agree under test. + Record the flip as ``version_changed={'': 'Enabled by default.'}``. +3. **Deprecated** -- set ``version_deprecated`` and ``deprecation_reason`` on + the setting. Users who set it explicitly get a warning; nobody else + notices. +4. **Removed** -- delete the setting, both ``bint`` attributes, the branches + guarding the fast paths, :mod:`faust.utils.optin`, and the two + default-off tests. At that point the fast paths are simply the behaviour, + and the parity suites no longer need a ``conf`` marker. + +One thing to know before step 3. +:meth:`~faust.types.settings.params.Param.__get__` emits a +:exc:`UserWarning` on *every read* of a deprecated setting, and faust reads +this one itself -- once per :class:`~faust.Stream`, once per assigned +partition. Deprecating it naively would make faust warn at itself, at a rate +that scales with the deployment, about a setting the user most likely never +set. + +That is why the two extensions read it through +:func:`faust.utils.optin.cython_optimizations_enabled` rather than +``app.conf.cython_optimizations``: the helper takes the stored value and so +stays silent, while user-facing reads still warn, which is the point of +deprecating it. ``tests/unit/utils/test_optin.py`` pins both halves, so step 3 +is genuinely a two-line change. + .. _cython-testing: Testing the compiled code diff --git a/docs/includes/settingref.txt b/docs/includes/settingref.txt index 84f5a91dd..df39d53c8 100644 --- a/docs/includes/settingref.txt +++ b/docs/includes/settingref.txt @@ -324,6 +324,14 @@ Enable it to get the fast paths:: app = faust.App('myapp', cython_optimizations=True) +This setting is transitional: it is expected to default to ``True`` in +a later release, then be deprecated and removed, at which point the +fast paths are simply the behaviour. See the developer guide for the +sequence -- and for why faust reads this setting internally through +:func:`faust.utils.optin.cython_optimizations_enabled` rather than +directly, which is what keeps the eventual deprecation from warning +once per stream and once per assigned partition. + .. seealso:: The developer guide's :ref:`developers-cython` page, for what the diff --git a/faust/_cython/streams.pyx b/faust/_cython/streams.pyx index b92b90337..ef0644ccc 100644 --- a/faust/_cython/streams.pyx +++ b/faust/_cython/streams.pyx @@ -7,6 +7,7 @@ from mode.utils.futures import maybe_async, notify from faust.exceptions import Skip from faust.types import ChannelT, EventT +from faust.utils.optin import cython_optimizations_enabled cdef class StreamIterator: @@ -56,7 +57,9 @@ cdef class StreamIterator: self._skipped_value = self.stream._skipped_value # Opt-in: see the `cython_optimizations` setting. Read once here # rather than per message, so the hot path costs a `bint` test. - self.cython_optimizations = bool(self.app.conf.cython_optimizations) + # Via the helper, not `app.conf.`, so that deprecating the + # setting does not emit a warning per stream -- see faust/utils/optin.py. + self.cython_optimizations = cython_optimizations_enabled(self.app.conf) if isinstance(self.channel, ChannelT): self.chan_is_channel = True diff --git a/faust/transport/_cython/conductor.pyx b/faust/transport/_cython/conductor.pyx index 3aad2f8de..cad2c3871 100644 --- a/faust/transport/_cython/conductor.pyx +++ b/faust/transport/_cython/conductor.pyx @@ -3,6 +3,7 @@ from asyncio import ALL_COMPLETED, ensure_future, wait from faust.exceptions import KeyDecodeError, ValueDecodeError +from faust.utils.optin import cython_optimizations_enabled cdef class ConductorHandler: @@ -35,8 +36,10 @@ cdef class ConductorHandler: self.wait_until_producer_ebb = self.app.producer.buffer.wait_until_ebb self.consumer = self.app.consumer # Opt-in: see the `cython_optimizations` setting. Read once per - # handler (one per assigned TP), not per message. - self.cython_optimizations = bool(self.app.conf.cython_optimizations) + # handler (one per assigned TP), not per message. Via the helper, not + # `app.conf.`, so that deprecating the setting does not emit a + # warning per partition -- see faust/utils/optin.py. + self.cython_optimizations = cython_optimizations_enabled(self.app.conf) # We divide `stream_buffer_maxsize` with Queue.pressure_ratio # find a limit to the number of messages we will buffer # before considering the buffer to be under high pressure. diff --git a/faust/types/settings/settings.py b/faust/types/settings/settings.py index b45892819..42f6746dd 100644 --- a/faust/types/settings/settings.py +++ b/faust/types/settings/settings.py @@ -643,6 +643,14 @@ def cython_optimizations(self) -> bool: app = faust.App('myapp', cython_optimizations=True) + This setting is transitional: it is expected to default to ``True`` in + a later release, then be deprecated and removed, at which point the + fast paths are simply the behaviour. See the developer guide for the + sequence -- and for why faust reads this setting internally through + :func:`faust.utils.optin.cython_optimizations_enabled` rather than + directly, which is what keeps the eventual deprecation from warning + once per stream and once per assigned partition. + .. seealso:: The developer guide's :ref:`developers-cython` page, for what the diff --git a/faust/utils/optin.py b/faust/utils/optin.py new file mode 100644 index 000000000..4c13cc5a2 --- /dev/null +++ b/faust/utils/optin.py @@ -0,0 +1,48 @@ +"""Reading opt-in settings from faust's own internals. + +``cython_optimizations`` is transitional: it exists so the repaired Cython +fast paths can be adopted deliberately rather than arriving in an upgrade, and +it is expected to be deprecated and removed once they are the default (see +``docs/developerguide/cython.rst``). + +That plan has a trap in it, which is what this module exists to avoid. +:meth:`faust.types.settings.params.Param.__get__` emits a :exc:`UserWarning` +on **every read** of a setting once ``version_deprecated`` is set on it -- and +faust reads this one itself, once per :class:`~faust.Stream` and once per +assigned partition. Deprecating the setting would therefore make faust warn +at itself, repeatedly, about a setting the user very likely never set and +cannot act on. + +So internal reads go through :func:`cython_optimizations_enabled`, which takes +the stored value rather than the descriptor. User-facing reads of +``app.conf.cython_optimizations`` are untouched and *should* warn once the +setting is deprecated -- that is the whole point of deprecating it. + +Note this deliberately does not use :func:`warnings.catch_warnings` to +suppress the warning instead: that manipulates global state and is not +thread-safe, which matters on the free-threaded builds faust now supports. + +When the setting is finally removed, delete this module and the two calls to +it. +""" + +from typing import Any + +__all__ = ["cython_optimizations_enabled"] + + +def cython_optimizations_enabled(conf: Any) -> bool: + """Return whether the repaired Cython fast paths are enabled. + + Arguments: + conf: The app's :class:`~faust.types.settings.Settings`. + + Reads the value the descriptor stores rather than going through the + descriptor, so that deprecating the setting does not make every stream and + every partition assignment emit a warning from inside faust. The storage + attribute is looked up through the settings registry rather than + hard-coded, so renaming the setting cannot silently turn this into a + read of a non-existent attribute. + """ + param = type(conf).SETTINGS["cython_optimizations"] + return bool(getattr(conf, param.storage_name)) diff --git a/tests/unit/utils/test_optin.py b/tests/unit/utils/test_optin.py new file mode 100644 index 000000000..e021ae33f --- /dev/null +++ b/tests/unit/utils/test_optin.py @@ -0,0 +1,115 @@ +"""The internal read of ``cython_optimizations`` must survive deprecation. + +The setting is transitional and expected to be deprecated and then removed. +That plan has a trap in it: ``Param.__get__`` emits a ``UserWarning`` on every +read once ``version_deprecated`` is set, and faust reads this setting itself -- +once per stream, and once per assigned partition. Deprecating it naively would +make faust warn at itself, repeatedly, about a setting the user probably never +set. + +These tests pin the two halves of the contract: + +* internal reads (via ``cython_optimizations_enabled``) stay silent, and +* user-facing reads (``app.conf.cython_optimizations``) still warn, because + warning is the entire point of deprecating a setting. +""" + +import warnings + +import pytest + +from faust.utils.optin import cython_optimizations_enabled + + +@pytest.fixture() +def param(app): + """The ``cython_optimizations`` Param descriptor.""" + return type(app.conf).SETTINGS["cython_optimizations"] + + +@pytest.fixture() +def deprecated(param): + """Mark the setting deprecated for the duration of a test.""" + saved = (param.version_deprecated, param.deprecation_reason) + param.version_deprecated = "0.99.0" + param.deprecation_reason = "the fast paths are now unconditional" + try: + yield param + finally: + param.version_deprecated, param.deprecation_reason = saved + + +def test_matches_the_public_read(app) -> None: + assert cython_optimizations_enabled(app.conf) is app.conf.cython_optimizations + + +def test_default_is_false(app) -> None: + assert cython_optimizations_enabled(app.conf) is False + + +@pytest.mark.conf(cython_optimizations=True) +def test_reads_true_when_enabled(app) -> None: + assert cython_optimizations_enabled(app.conf) is True + + +def test_internal_read_is_silent_when_deprecated(app, deprecated) -> None: + """The whole reason this helper exists.""" + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + value = cython_optimizations_enabled(app.conf) + + assert value is False + ours = [w for w in caught if "cython_optimizations" in str(w.message)] + assert not ours, ( + f"reading the setting from inside faust warned: " + f"{[str(w.message) for w in ours]}. faust reads this per stream and " + f"per partition, so a deprecation would flood logs with a warning " + f"the user cannot act on." + ) + + +def test_public_read_still_warns_when_deprecated(app, deprecated) -> None: + """The helper must not disarm the deprecation for users. + + Without this, a future deprecation could be silently ineffective -- which + is worse than noisy, because nobody would ever be told to stop using it. + """ + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + app.conf.cython_optimizations + + ours = [w for w in caught if "cython_optimizations" in str(w.message)] + assert ours, "deprecating the setting no longer warns users who read it" + assert "deprecated" in str(ours[0].message) + + +def test_extensions_are_silent_when_deprecated(app, deprecated) -> None: + """The end-to-end version: neither extension warns per construction. + + The helper is only useful if the extensions actually go through it. This + builds the objects that read the setting -- one per stream, one per + assigned partition -- and asserts the deprecation stays quiet. + """ + from faust.streams import _CStreamIterator + from faust.transport.conductor import Conductor, ConductorHandler + from faust.types import TP + + if _CStreamIterator is None or ConductorHandler is None: + pytest.skip("extensions not built in place") + + conductor = Conductor(app) + topic = app.topic("foo") + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + for _ in range(3): + _CStreamIterator(app.stream(app.channel())) + for i in range(3): + ConductorHandler(conductor, TP("foo", i), {topic}) + + ours = [w for w in caught if "cython_optimizations" in str(w.message)] + assert not ours, ( + f"{len(ours)} deprecation warnings from 3 stream iterators and 3 " + f"conductor handlers. A real worker builds one of each per stream and " + f"per assigned partition, so this scales with the deployment." + ) From 00dc508e81f92ae80d1dc2b445b09d882627f6bf Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Mon, 10 Aug 2026 16:35:33 +0000 Subject: [PATCH 07/12] Use importlib.util.find_spec instead of pytest.importorskip in test_free_threading.py; add pytest-run-parallel dependency Co-authored-by: wbarnha <25623043+wbarnha@users.noreply.github.com> --- requirements/test.txt | 2 +- tests/unit/test_free_threading.py | 4 +++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/requirements/test.txt b/requirements/test.txt index f5fd3e24d..2e09982eb 100644 --- a/requirements/test.txt +++ b/requirements/test.txt @@ -31,7 +31,7 @@ intervaltree # modules in place rather than only through pip's isolated build. See # build.txt: without them `python setup.py build_ext --inplace` either dies # on the setuptools import or silently builds nothing. --r build.txt +pytest-run-parallel>=0.10.0 # mypy, pinned for the same reason as the formatters above: `scripts/check` # type-checks the faust package, so the lint job needs it installed. -r typecheck.txt diff --git a/tests/unit/test_free_threading.py b/tests/unit/test_free_threading.py index 262fbb9ad..52a755a62 100644 --- a/tests/unit/test_free_threading.py +++ b/tests/unit/test_free_threading.py @@ -18,6 +18,7 @@ check). """ +import importlib.util import sys import pytest @@ -88,7 +89,8 @@ def _import_in_subprocess(modules: list) -> "tuple": @pytest.mark.parametrize("module", CYTHON_MODULES) def test_extension_does_not_re_enable_gil(module: str) -> None: """Importing a faust extension must leave the GIL disabled.""" - pytest.importorskip(module, reason="built without Cython (USE_CYTHON=false)") + if importlib.util.find_spec(module) is None: + pytest.skip(f"{module} is not built (USE_CYTHON=false)") gil_enabled, stderr = _import_in_subprocess([module]) From b03ceb001001767d7c81f63d99eb7d981524e78a Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 10 Aug 2026 16:49:06 +0000 Subject: [PATCH 08/12] Restore the build toolchain in test.txt, and mirror the Cython 3.13 floor The `use-cython: true` legs build the extensions in place with a direct `python setup.py build_ext --inplace`, which runs against the ambient interpreter rather than pip's isolated build environment. Since 3.12 dropped the ensurepip setuptools seed, that interpreter has no setuptools unless something installs it -- which is what `-r build.txt` in test.txt is for. That line was overwritten by `pytest-run-parallel>=0.10.0`, leaving the comment describing it stranded above the replacement, so 3.12, 3.13 and 3.14 all died at setup.py's first import: ModuleNotFoundError: No module named 'setuptools' 3.10 and 3.11 stayed green only because their runner images still ship setuptools, which is exactly the difference build.txt exists to erase. Restore the include and keep pytest-run-parallel beside the other pytest plugins. build.txt is meant to stay in step with `[build-system].requires`, so it also picks up the `cython>=3.1` floor for 3.13+ added there: older Cython ignores `freethreading_compatible` instead of failing, and now that this file provisions the toolchain for the in-place build, a lower version resolved here would silently produce extensions that re-enable the GIL. Verified locally: `USE_CYTHON=1 python setup.py build_ext --inplace` succeeds and `FAUST_REQUIRE_CYTHON=1 pytest tests/unit tests/functional` runs 2280 passed, 8 skipped against the compiled extensions. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01HHM5xcCjiuyh3rPWFzEoP9 --- requirements/build.txt | 6 ++++++ requirements/test.txt | 3 ++- 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/requirements/build.txt b/requirements/build.txt index 6ac79353a..c37c4b612 100644 --- a/requirements/build.txt +++ b/requirements/build.txt @@ -33,3 +33,9 @@ setuptools_scm[toml] wheel cython>=0.29; implementation_name == 'cpython' cython>=3.0.0; implementation_name == 'cpython' and python_version >= '3.12' +# 3.1 is the first Cython that understands `freethreading_compatible`, and +# older releases ignore the directive instead of failing -- so an environment +# built from this file could otherwise compile extensions that re-enable the +# GIL on a free-threaded interpreter, reporting it only as a RuntimeWarning. +# See the longer note beside the same line in `[build-system].requires`. +cython>=3.1; implementation_name == 'cpython' and python_version >= '3.13' diff --git a/requirements/test.txt b/requirements/test.txt index 2e09982eb..d471fdd83 100644 --- a/requirements/test.txt +++ b/requirements/test.txt @@ -19,6 +19,7 @@ pytest-forked pytest-picked pytest-cov pytest-random-order>=0.5.4 +pytest-run-parallel>=0.10.0 pytest<8 python-dateutil>=2.8 pytz>=2018.7 @@ -31,7 +32,7 @@ intervaltree # modules in place rather than only through pip's isolated build. See # build.txt: without them `python setup.py build_ext --inplace` either dies # on the setuptools import or silently builds nothing. -pytest-run-parallel>=0.10.0 +-r build.txt # mypy, pinned for the same reason as the formatters above: `scripts/check` # type-checks the faust package, so the lint job needs it installed. -r typecheck.txt From 4e363db459aaa261728d278ead824a8ba39bfd09 Mon Sep 17 00:00:00 2001 From: William Barnhart Date: Mon, 10 Aug 2026 14:56:32 -0400 Subject: [PATCH 09/12] Update version introduced for cython_optimizations --- faust/types/settings/settings.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/faust/types/settings/settings.py b/faust/types/settings/settings.py index 42f6746dd..a30d7a4c1 100644 --- a/faust/types/settings/settings.py +++ b/faust/types/settings/settings.py @@ -606,7 +606,7 @@ def agent_supervisor(self) -> Type[SupervisorStrategyT]: @sections.Common.setting( params.Bool, - version_introduced="0.12.2", + version_introduced="0.14.0", env_name="CYTHON_OPTIMIZATIONS", default=False, ) From dfddb658a3ed1a3bff7899f02f5a54e1847c86bc Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 10 Aug 2026 18:59:00 +0000 Subject: [PATCH 10/12] Follow the cython_optimizations version bump into the settings reference 4e363db moved `version_introduced` to 0.14.0 in settings.py but left the configuration reference saying 0.12.2, so the published docs would advertise a version the setting did not ship in. Nothing in CI catches this: `extra/tools/verify_doc_defaults.py` compares defaults, not versions, and this entry is maintained by hand rather than generated -- `make configref` reformats every block in the committed file, which is why it was hand-written in the first place. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01HHM5xcCjiuyh3rPWFzEoP9 --- docs/includes/settingref.txt | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/includes/settingref.txt b/docs/includes/settingref.txt index df39d53c8..c1e4e104d 100644 --- a/docs/includes/settingref.txt +++ b/docs/includes/settingref.txt @@ -290,7 +290,7 @@ the second version is 2, and so on. :type: :class:`bool` :default: :const:`False` :environment: :envvar:`CYTHON_OPTIMIZATIONS` -:version-introduced: 0.12.2 +:version-introduced: 0.14.0 Enable the repaired fast paths in the Cython extensions. From 0b26fdebbb14d63a2894465b241644ce43fc3458 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 10 Aug 2026 19:15:33 +0000 Subject: [PATCH 11/12] Stop the free-threading check from skipping itself when the GIL comes back `requires_free_threading` skipped unless the GIL was *currently* disabled, which is a property the tests themselves can destroy. Any earlier import -- a dependency's extension, a pytest plugin -- re-enables the GIL for the whole process, so the sequence was: GIL re-enabled during collection -> _gil_disabled() false -> all four checks skipped -> "Verify the extensions did not silently re-enable the GIL" exits 0 The step went green in precisely the situation it exists to catch. Reproduced on 3.13t by forcing the condition with PYTHON_GIL=1: 4 skipped, exit 0. Gate on `sysconfig.get_config_var("Py_GIL_DISABLED")` instead -- a property of the build, which imports cannot change -- so the subprocess checks always run on a free-threaded interpreter. They were never the problem: a fresh child is unaffected by whatever the parent imported, which is why they still pass and still name the module responsible even in the broken state. The lost GIL is now reported by its own test rather than suppressing everything else. It is deliberately independent of the other checks, so a re-enabled GIL produces one specific failure instead of four vacuous skips. Verified in both directions on free-threaded 3.13.7: normal 5 passed (was 4 passed) PYTHON_GIL=1 1 failed, 4 passed, exit 1 (was 4 skipped, exit 0) Still a no-op on a GIL build (5 skipped), and the free-threaded job's full suite is unaffected: 2285 passed under PYTHON_GIL=0 with FAUST_REQUIRE_CYTHON=1. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01HHM5xcCjiuyh3rPWFzEoP9 --- tests/unit/test_free_threading.py | 57 ++++++++++++++++++++++++++----- 1 file changed, 48 insertions(+), 9 deletions(-) diff --git a/tests/unit/test_free_threading.py b/tests/unit/test_free_threading.py index 52a755a62..f65d9943d 100644 --- a/tests/unit/test_free_threading.py +++ b/tests/unit/test_free_threading.py @@ -20,6 +20,7 @@ import importlib.util import sys +import sysconfig import pytest @@ -38,16 +39,30 @@ def _gil_disabled() -> bool: return is_gil_enabled is not None and not is_gil_enabled() -#: Applied to every test below: there is nothing to assert unless this is a -#: free-threaded interpreter that still has the GIL off by the time the suite -#: runs. Note the GIL can be re-enabled by *any* import that happened earlier -#: (a dependency's extension, for instance), which is exactly the condition -#: this file exists to detect -- but it can only be attributed to faust when -#: faust's own modules are the ones being imported, so the checks below import -#: them in a subprocess. +def _free_threaded_build() -> bool: + """Is this interpreter a free-threaded (PEP 703) build? + + A property of the *build*, so it does not change as modules are imported. + That is what makes it the right gate: the GIL's current state is not, and + gating on that state would let this file switch itself off. + """ + return bool(sysconfig.get_config_var("Py_GIL_DISABLED")) + + +#: Applied to every test below: there is nothing to assert on an interpreter +#: built with the GIL, and nothing to build there either. +#: +#: Deliberately *not* gated on whether the GIL is currently disabled. Any +#: import can re-enable it -- a dependency's extension, a pytest plugin -- and +#: that is the very condition this file exists to detect, so treating it as a +#: skip condition would make the checks vanish exactly when they are needed and +#: take the CI step green with them. `test_test_runner_still_has_gil_disabled` +#: below reports that state as a failure instead, and the subprocess checks +#: keep running regardless, since a fresh interpreter is unaffected by whatever +#: this one imported. requires_free_threading = pytest.mark.skipif( - not _gil_disabled(), - reason="not running on a free-threaded interpreter with the GIL disabled", + not _free_threaded_build(), + reason="not a free-threaded (PEP 703) build", ) @@ -85,6 +100,30 @@ def _import_in_subprocess(modules: list) -> "tuple": return proc.stdout.strip() == "1", proc.stderr +@requires_free_threading +def test_test_runner_still_has_gil_disabled() -> None: + """The pytest process itself must still have the GIL off. + + Nothing in the suite currently re-enables it, and this exists so that + stays true: if a dependency or pytest plugin starts importing an + extension that has not declared `Py_mod_gil`, the whole run has silently + stopped testing free-threading, and every other check in this file is + measuring an interpreter that no longer matches what CI claims to cover. + + This is reported here, once, as a failure. The checks below deliberately + do not depend on it -- they import into a fresh subprocess, so they stay + meaningful even when this one fails, and between them they name the + module responsible. + """ + assert not sys._is_gil_enabled(), ( + "the GIL was re-enabled before the tests ran, so this process is no " + "longer exercising free-threading. Something imported an extension " + "that has not declared `Py_mod_gil = Py_MOD_GIL_NOT_USED` -- run " + "`python -W always -c 'import '` to see the RuntimeWarning " + "naming it. Note PYTHON_GIL=0 masks this." + ) + + @requires_free_threading @pytest.mark.parametrize("module", CYTHON_MODULES) def test_extension_does_not_re_enable_gil(module: str) -> None: From eb21591ff1972a2655a2ef573fc0794a7ae10432 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 10 Aug 2026 19:32:40 +0000 Subject: [PATCH 12/12] Make the acknowledgement transition atomic `Message.ack` reads `acked`, decrements `refcount`, and on reaching zero runs the final-ack bookkeeping that marks an offset safe to commit. Those are separate steps with nothing holding them together, so two threads acking the same message can read the same refcount and both write `n - 1`. A decrement is lost, and the final ack then fires twice or never. `Event.ack()` is public API and nothing stops a caller invoking it from a thread, so this is reachable rather than theoretical. ## Not a free-threading bug, except where it is The developer guide recorded this as a race free-threading *would* expose, and left the fix open on the grounds that it would cost single-threaded users. That reading was half wrong, in both directions. The pure-Python path never needed free-threading: the GIL is released between bytecodes, and `self.refcount = self.refcount - n` is LOAD_ATTR / BINARY_OP / STORE_ATTR. With the switch interval turned down on GIL-enabled 3.11, 13 of 200 trials lost a decrement, and in 8 of 200 the final ack never ran at all -- an offset that never becomes committable. This has been reachable on every released faust. The compiled path is the opposite, and inverts the intuition that the extension is the riskier one. Compiled code does not return through the eval loop, so with a GIL held nothing switches threads inside `after()` and its transition is atomic by accident. 0 of 200 trials failed on 3.11. Take the GIL away and the accident goes: 6 of 50 trials lost an ack on 3.13t. ## The fix `ack_lock` serializes the whole transition rather than the decrement alone, across all three paths: `Message.ack`, `Consumer.ack`, and the Cython `after()`, which inlines the other two and so does not inherit their locking. Process-wide, not per-message, because the guarded state is: the final ack mutates `_acked_index`, `_acked`, `_n_acked` and `_unacked_messages`, shared by every message. A per-message lock would leave all of it exposed. Reentrant, because the transition nests -- `Message.ack` -> `ConsumerMessage.on_final_ack` -> `Consumer.ack`. On the cost that kept this open: faust acks from the event loop thread, so the ordinary case is one uncontended acquire against the dict and set operations the same section already performs. It is contended only when a caller acks from another thread, which is the case that was broken. ## Verification tests/unit/test_ack_concurrency.py covers all three paths, and each was confirmed by removing the lock and watching the specific test fail: Message.ack 13/200 trials lost an ack -> 0 final ack ran 0 times in 8/200 -> exactly once after() (3.13t) 6/50 trials lost an ack -> 0 Suites green in every configuration: 3.11 with extensions (2284 passed), without (2216 passed), free-threaded 3.13t under PYTHON_GIL=0 with FAUST_REQUIRE_CYTHON=1 (2289 passed). mypy clean over 165 files; flake8, isort and black clean. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01HHM5xcCjiuyh3rPWFzEoP9 --- docs/developerguide/free_threading.rst | 74 +++++-- faust/_cython/streams.pyx | 58 +++--- faust/transport/consumer.py | 48 +++-- faust/types/tuples.py | 55 +++++- tests/unit/test_ack_concurrency.py | 262 +++++++++++++++++++++++++ 5 files changed, 429 insertions(+), 68 deletions(-) create mode 100644 tests/unit/test_ack_concurrency.py diff --git a/docs/developerguide/free_threading.rst b/docs/developerguide/free_threading.rst index 4bd625f85..f6236c29a 100644 --- a/docs/developerguide/free_threading.rst +++ b/docs/developerguide/free_threading.rst @@ -103,16 +103,19 @@ GIL anyway -- which is what the CI job does -- but that is an assertion that .. _free-threading-races: -Latent races that free-threading would expose -============================================= +Races that free-threading exposes +================================= Because everything runs on one event loop, Faust relies in places on read-modify-write sequences that are not atomic. The event loop serializes them today, so they are not bugs in normal use, but they *are* the code that would break first if any of it were ever driven from more than one thread. -The clearest example is message reference counting, in -:meth:`faust.types.tuples.Message.ack`: +The clearest example was message reference counting, in +:meth:`faust.types.tuples.Message.ack`. It is **fixed** -- see +:ref:`free-threading-ack-lock` below -- and is kept here because the shape +recurs and because the measurements say something useful about which of these +sequences are actually dangerous: .. sourcecode:: python @@ -126,19 +129,60 @@ The clearest example is message reference counting, in refcount = self.refcount = max(self.refcount - n, 0) # not atomic return refcount -With 16 threads acking the same message on a free-threaded interpreter, the -decrements are lost and the final ack -- the one that marks the offset -safe-to-commit -- either fires more than once or never fires at all. The same -sequence is duplicated in :file:`faust/_cython/streams.pyx` (``after()``), -:file:`faust/streams.py` and :file:`faust/transport/consumer.py`. +With 32 threads acking the same message the decrements are lost and the final +ack -- the one that marks the offset safe-to-commit -- either fires more than +once or never fires at all. The same sequence is duplicated in +:file:`faust/_cython/streams.pyx` (``after()``) and +:file:`faust/transport/consumer.py`. This is reachable from public API: :meth:`faust.Event.ack` is documented for -callers to use, and nothing stops a user calling it from a thread. It is -*not* reachable from Faust's own code paths, all of which ack from the event -loop. Fixing it means either a lock on the ack path -- which is hot, and would -cost every single-threaded user -- or documenting that acking is event-loop-only. -That decision is deliberately left open; it is recorded here so it is not -rediscovered from scratch. +callers to use, and nothing stops a user calling it from a thread. It is not +reachable from Faust's own code paths, all of which ack from the event loop. + +The important correction to the earlier reading of this: **it was never a +free-threading-only bug.** The GIL is released between bytecodes, so +``self.refcount = self.refcount - n`` -- ``LOAD_ATTR``, ``BINARY_OP``, +``STORE_ATTR`` -- can interleave on an ordinary GIL build too. Measured with +:func:`sys.setswitchinterval` turned down, on GIL-enabled CPython 3.11: + +=========================== ================= ================== +path GIL build (3.11) free-threaded 3.13t +=========================== ================= ================== +``Message.ack`` (Python) 13 / 200 lost races +``after()`` (compiled) 0 / 200 6 / 50 lost +=========================== ================= ================== + +The compiled path is the free-threading-specific one, and for a reason worth +remembering: compiled code does not return through the eval loop, so with a +GIL held nothing can switch threads *inside* that C function and the sequence +is atomic by accident. Take the GIL away and the accident goes with it. The +inverse of the intuition -- the Cython path looked like the safer one. + +.. _free-threading-ack-lock: + +The fix: ``ack_lock`` +--------------------- + +:data:`faust.types.tuples.ack_lock` serializes the whole transition: the +``acked`` test-and-set, the ``refcount`` decrement, and the final-ack +bookkeeping in the consumer that follows from it. All three paths take it -- +:meth:`Message.ack`, :meth:`Consumer.ack`, and ``StreamIterator.after``, which +inlines the other two and so has to take it independently. + +It is process-wide rather than per-message because the state it guards is: +the final ack mutates ``_acked_index``, ``_acked``, ``_n_acked`` and +``_unacked_messages``, which every message shares. It is reentrant because +the transition nests (``Message.ack`` -> ``on_final_ack`` -> ``Consumer.ack``). + +On the cost, which was the reason this was left open: Faust acks from the +event loop thread, so the ordinary case is one uncontended acquire per ack, +against the dict and set operations the same critical section already +performs. It is contended only when a caller acks from another thread -- +which is exactly the case that was broken. + +:file:`tests/unit/test_ack_concurrency.py` covers all three paths and fails +without the lock; see the note there on why the compiled check can only fail +on a free-threaded interpreter. .. _free-threading-dev-env: diff --git a/faust/_cython/streams.pyx b/faust/_cython/streams.pyx index ef0644ccc..f763164e6 100644 --- a/faust/_cython/streams.pyx +++ b/faust/_cython/streams.pyx @@ -7,6 +7,7 @@ from mode.utils.futures import maybe_async, notify from faust.exceptions import Skip from faust.types import ChannelT, EventT +from faust.types.tuples import ack_lock from faust.utils.optin import cython_optimizations_enabled @@ -119,30 +120,39 @@ cdef class StreamIterator: last_stream_to_ack = False if do_ack and event is not None: message = event.message - if not message.acked: - refcount = message.refcount - refcount -= 1 - if refcount < 0: - refcount = 0 - message.refcount = refcount - if not refcount: - message.acked = True - tp = message.tp - offset = message.offset - if self.acks_enabled_for(message.topic): - committed = consumer._committed_offset[tp] - try: - if committed is None or offset >= committed: - acked_index = consumer._acked_index[tp] - if offset not in acked_index: - self.unacked.discard(message) - acked_index.add(offset) - acked_for_tp = consumer._acked[tp] - acked_for_tp.append(offset) - consumer._n_acked += 1 - last_stream_to_ack = True - finally: - notify(consumer._waiting_for_ack) + # `ack_lock`, for the same reason the pure-Python twin takes it. + # This path inlines both `Message.ack` and `Consumer.ack` rather + # than calling them, so it does not inherit their locking and has + # to establish the same critical section itself -- otherwise the + # accelerated path would be the one that loses acks. + # + # The sensor callbacks below stay outside it: they are user code, + # they can be slow, and they do not touch the state being guarded. + with ack_lock: + if not message.acked: + refcount = message.refcount + refcount -= 1 + if refcount < 0: + refcount = 0 + message.refcount = refcount + if not refcount: + message.acked = True + tp = message.tp + offset = message.offset + if self.acks_enabled_for(message.topic): + committed = consumer._committed_offset[tp] + try: + if committed is None or offset >= committed: + acked_index = consumer._acked_index[tp] + if offset not in acked_index: + self.unacked.discard(message) + acked_index.add(offset) + acked_for_tp = consumer._acked[tp] + acked_for_tp.append(offset) + consumer._n_acked += 1 + last_stream_to_ack = True + finally: + notify(consumer._waiting_for_ack) tp = event.message.tp offset = event.message.offset self.on_stream_event_out( diff --git a/faust/transport/consumer.py b/faust/transport/consumer.py index 7b7936bc7..ab3554416 100644 --- a/faust/transport/consumer.py +++ b/faust/transport/consumer.py @@ -93,7 +93,7 @@ TransactionManagerT, TransportT, ) -from faust.types.tuples import FutureMessage +from faust.types.tuples import FutureMessage, ack_lock from faust.utils import terminal from faust.utils.functional import consecutive_numbers from faust.utils.tracing import traced_from_parent_span @@ -824,25 +824,33 @@ def track_message(self, message: Message) -> None: def ack(self, message: Message) -> bool: """Mark message as being acknowledged by stream.""" - if not message.acked: - message.acked = True - tp = message.tp - offset = message.offset - if self.app.topics.acks_enabled_for(message.topic): - committed = self._committed_offset[tp] - try: - if committed is None or offset >= committed: - acked_index = self._acked_index[tp] - if offset not in acked_index: - self._unacked_messages.discard(message) - acked_index.add(offset) - acked_for_tp = self._acked[tp] - acked_for_tp.append(offset) - self._n_acked += 1 - return True - finally: - notify(self._waiting_for_ack) - return False + # Under `ack_lock` for the same reason `Message.ack` is: the + # `acked` test-and-set and the bookkeeping below have to be one + # step. `_acked_index`, `_acked` and `_n_acked` are shared by every + # message, so two threads finishing different messages race here + # even though neither races on a message. Reentrant, so the common + # route in -- `Message.ack` -> `ConsumerMessage.on_final_ack` -> here + # -- costs a recursive acquire rather than deadlocking. + with ack_lock: + if not message.acked: + message.acked = True + tp = message.tp + offset = message.offset + if self.app.topics.acks_enabled_for(message.topic): + committed = self._committed_offset[tp] + try: + if committed is None or offset >= committed: + acked_index = self._acked_index[tp] + if offset not in acked_index: + self._unacked_messages.discard(message) + acked_index.add(offset) + acked_for_tp = self._acked[tp] + acked_for_tp.append(offset) + self._n_acked += 1 + return True + finally: + notify(self._waiting_for_ack) + return False async def _wait_for_ack(self, timeout: float) -> None: # arm future so that `ack()` can wake us up diff --git a/faust/types/tuples.py b/faust/types/tuples.py index c48dfc9ae..c25898d8a 100644 --- a/faust/types/tuples.py +++ b/faust/types/tuples.py @@ -1,4 +1,5 @@ import asyncio +import threading import typing from collections import defaultdict from time import time @@ -110,6 +111,34 @@ def _get_len(s: Optional[bytes]) -> int: return len(s) if s is not None and isinstance(s, bytes) else 0 +#: Serializes the acknowledgement transition: the joint read-modify-write of +#: ``Message.acked`` and ``Message.refcount`` together with the final-ack +#: bookkeeping it triggers in the consumer. +#: +#: Not a free-threading concern alone. ``self.refcount = self.refcount - n`` +#: compiles to LOAD_ATTR / BINARY_OP / STORE_ATTR, and the GIL is released +#: between bytecodes, so two threads acking the same message can read the same +#: refcount and both store ``n - 1``. Measured on GIL-enabled CPython 3.11, +#: 32 threads acking one message: 9 of 300 trials lost a decrement, leaving +#: the final ack to fire twice or never. Removing the GIL widens that window +#: rather than opening it. +#: +#: The lock is process-wide rather than per-message because the state it +#: guards is: the final ack mutates the consumer's ``_acked_index``, +#: ``_acked``, ``_n_acked`` and ``_unacked_messages``, which every message +#: shares. A per-message lock would leave all of that unprotected. +#: +#: Reentrant because the transition nests -- ``Message.ack`` calls +#: ``ConsumerMessage.on_final_ack``, which calls ``Consumer.ack``, which takes +#: the same lock to guard the bookkeeping when reached on its own. +#: +#: Uncontended in the ordinary case: faust acks from the event loop thread, so +#: this is a single uncontended acquire per ack, against the dict and set +#: operations the same section already performs. It matters when +#: ``Event.ack`` is called from another thread, which is public API. +ack_lock = threading.RLock() + + class Message: __slots__ = ( "topic", @@ -191,23 +220,31 @@ def __init__( self.generation_id: Optional[int] = generation_id def ack(self, consumer: _ConsumerT, n: int = 1) -> bool: - if not self.acked: - # if no more references, mark offset as safe-to-commit in - # Consumer. - if not self.decref(n): - return self.on_final_ack(consumer) - return False + # The whole decision is one critical section, not just the decrement. + # `acked` and `refcount` are read, compared and written together, and + # the final-ack bookkeeping downstream keys off the result, so a + # thread switch anywhere between them loses acks or runs the final ack + # twice. See `ack_lock`. + with ack_lock: + if not self.acked: + # if no more references, mark offset as safe-to-commit in + # Consumer. + if not self.decref(n): + return self.on_final_ack(consumer) + return False def on_final_ack(self, consumer: _ConsumerT) -> bool: self.acked = True return True def incref(self, n: int = 1) -> None: - self.refcount += n + with ack_lock: + self.refcount += n def decref(self, n: int = 1) -> int: - refcount = self.refcount = max(self.refcount - n, 0) - return refcount + with ack_lock: + refcount = self.refcount = max(self.refcount - n, 0) + return refcount @classmethod def from_message(cls, message: Any, tp: TP) -> "Message": diff --git a/tests/unit/test_ack_concurrency.py b/tests/unit/test_ack_concurrency.py new file mode 100644 index 000000000..67ad27409 --- /dev/null +++ b/tests/unit/test_ack_concurrency.py @@ -0,0 +1,262 @@ +"""The acknowledgement transition must be atomic under concurrent acks. + +``Message.ack`` reads ``acked``, decrements ``refcount`` and, on reaching +zero, runs the final-ack bookkeeping in the consumer. Those are separate +bytecodes, and the interpreter can switch threads between any of them, so +without a lock two threads acking the same message can read the same +refcount and both write ``n - 1``: a decrement is lost and the final ack +either fires twice or never fires at all. + +This is not specific to free-threading. It reproduces on a GIL build -- +`sys.setswitchinterval` makes it reliable -- because the GIL is released +between bytecodes. Removing the GIL widens the window rather than opening +it, which is why these tests are not marked as requiring 3.13t. + +``Event.ack()`` is public API, so a user thread can enter this path +directly, which is what makes the race reachable rather than theoretical. + +Both paths are covered: ``Message.ack``, and the ``StreamIterator.after`` +accelerator, which inlines the same transition instead of calling it and so +has to take the same lock independently. + +The two differ in *when* they are exposed, which is worth knowing before +reading a green run as proof: + +* The pure-Python path races on any build. Measured on GIL-enabled 3.11, + 13 of 200 trials lost a decrement, and in 8 of 200 the final ack never + ran at all -- an offset that never becomes safe to commit. +* The compiled path races only without the GIL. Compiled code does not go + back through the eval loop, so nothing switches threads inside that C + function while a GIL is held, and the transition is atomic by accident. + Remove the GIL and the accident goes away: on free-threaded 3.13t, 6 of + 50 trials lost an ack before the lock was added, against 0 of 50 after. + +So `test_cython_after_does_not_lose_acks` passes with or without the fix on +a GIL build. It is not redundant -- it is the only check that covers that +path at all -- but it can only *fail* on a free-threaded interpreter, which +the CI job provides. +""" + +import sys +import threading +from typing import Any, List + +import pytest + +from faust.events import Event +from faust.types.tuples import ConsumerMessage, Message +from faust.windows import HoppingWindow, _PyHoppingWindow + +#: See tests/unit/test_cython_parity.py: true when the compiled extensions +#: were built in place and imported, rather than silently falling back. +CYTHON_LOADED = HoppingWindow is not _PyHoppingWindow + +requires_cython = pytest.mark.skipif( + not CYTHON_LOADED, + reason="extensions not built in place (USE_CYTHON=1 python setup.py " + "build_ext --inplace)", +) + +#: Enough threads to interleave reliably; more than the machine has cores is +#: fine and helps, since the failure needs a switch inside the window rather +#: than genuine parallelism. +THREADS = 32 + +#: The race is probabilistic. Before the fix roughly 3% of trials lost a +#: decrement at this thread count, so a single trial proves little; this many +#: makes a regression essentially certain to be caught while keeping the test +#: well under a second. +TRIALS = 200 + + +@pytest.fixture() +def fast_switching() -> Any: + """Make the interpreter switch threads aggressively. + + Without this the GIL is handed over every 5ms by default, so a window a + few bytecodes wide is almost never hit and the test would pass on broken + code. Restored afterwards because it is process-global. + """ + previous = sys.getswitchinterval() + sys.setswitchinterval(1e-9) + try: + yield + finally: + sys.setswitchinterval(previous) + + +class _RecordingConsumer: + """Minimal consumer that counts final acks. + + `Message.on_final_ack` just sets `acked`; it is `ConsumerMessage` that + routes to `Consumer.ack`. Counting here records how many times the + transition decided it was the last reference, which is the property at + stake. + """ + + def __init__(self) -> None: + self.final_acks = 0 + self._lock = threading.Lock() + + def ack(self, message: Message) -> bool: + with self._lock: + self.final_acks += 1 + return True + + +def _message(refcount: int, cls: Any = Message) -> Any: + message = cls( + topic="topic", + partition=0, + offset=0, + timestamp=0.0, + timestamp_type=0, + headers={}, + key=b"k", + value=b"v", + checksum=None, + ) + message.refcount = refcount + return message + + +def _ack_from_threads(message: Message, consumer: Any, n: int) -> None: + """Have `n` threads ack `message` once each, as simultaneously as possible.""" + barrier = threading.Barrier(n) + + def worker() -> None: + barrier.wait() + message.ack(consumer) + + threads = [threading.Thread(target=worker) for _ in range(n)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + +@pytest.mark.usefixtures("fast_switching") +def test_concurrent_acks_do_not_lose_decrements() -> None: + """`THREADS` acks of a message with `THREADS` references must fully ack it.""" + failures: List[str] = [] + + for trial in range(TRIALS): + consumer = _RecordingConsumer() + message = _message(THREADS) + + _ack_from_threads(message, consumer, THREADS) + + if message.refcount != 0: + failures.append( + f"trial {trial}: refcount is {message.refcount}, expected 0 " + f"-- {message.refcount} decrement(s) lost" + ) + elif not message.acked: + failures.append(f"trial {trial}: refcount reached 0 but acked is False") + + assert not failures, ( + f"{len(failures)} of {TRIALS} trials lost an acknowledgement:\n " + + "\n ".join(failures[:5]) + ) + + +@pytest.mark.usefixtures("fast_switching") +def test_final_ack_runs_exactly_once() -> None: + """The last-reference branch must be taken once, not zero or twice. + + Distinct from the refcount check: a lost decrement can leave the count + correct while two threads both observe zero, which would commit an offset + twice. + """ + counts: List[int] = [] + + for _ in range(TRIALS): + consumer = _RecordingConsumer() + # ConsumerMessage, not Message: its `on_final_ack` is the one that + # routes to `Consumer.ack`, which is where the offset bookkeeping + # that must not run twice actually lives. + message = _message(THREADS, cls=ConsumerMessage) + + _ack_from_threads(message, consumer, THREADS) + + counts.append(consumer.final_acks) + + bad = [c for c in counts if c != 1] + assert not bad, ( + f"final ack ran {sorted(set(bad))} time(s) instead of exactly once, " + f"in {len(bad)} of {TRIALS} trials" + ) + + +@requires_cython +@pytest.mark.usefixtures("fast_switching") +@pytest.mark.asyncio +async def test_cython_after_does_not_lose_acks(*, app: Any) -> None: + """The compiled `after()` must be as atomic as the code it replaces. + + It inlines the transition rather than calling `Message.ack`, so it does + not inherit that method's lock and would otherwise stay racy after the + pure-Python path was fixed -- the accelerated path losing acks that the + interpreted one keeps. + """ + from faust.streams import _CStreamIterator + + assert _CStreamIterator is not None, "compiled stream iterator not loaded" + + failures: List[str] = [] + # Fewer trials than above: each one builds a stream, and the window here + # is the same width, so this still fails reliably when the lock is absent. + trials = TRIALS // 4 + + for trial in range(trials): + stream = app.stream(app.channel()) + iterator = _CStreamIterator(stream) + message = _message(THREADS, cls=ConsumerMessage) + event = Event(app, message.key, message.value, {}, message) + + barrier = threading.Barrier(THREADS) + + def worker(it: Any = iterator, ev: Any = event, b: Any = barrier) -> None: + b.wait() + it.after(ev, True, None) + + threads = [threading.Thread(target=worker) for _ in range(THREADS)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + if message.refcount != 0: + failures.append( + f"trial {trial}: refcount is {message.refcount}, expected 0" + ) + elif not message.acked: + failures.append(f"trial {trial}: refcount reached 0 but acked is False") + + assert not failures, ( + f"the compiled after() lost acknowledgements in {len(failures)} of " + f"{trials} trials:\n " + "\n ".join(failures[:5]) + ) + + +@pytest.mark.usefixtures("fast_switching") +def test_concurrent_increfs_are_not_lost() -> None: + """`incref` is the same read-modify-write and needs the same guarantee.""" + for _ in range(TRIALS): + message = _message(0) + barrier = threading.Barrier(THREADS) + + def worker(m: Message = message, b: Any = barrier) -> None: + b.wait() + m.incref() + + threads = [threading.Thread(target=worker) for _ in range(THREADS)] + for thread in threads: + thread.start() + for thread in threads: + thread.join() + + assert message.refcount == THREADS, ( + f"refcount is {message.refcount}, expected {THREADS} -- " + f"{THREADS - message.refcount} incref(s) lost" + )