diff --git a/.github/workflows/python-package.yml b/.github/workflows/python-package.yml index 79fb3b54a..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 @@ -138,6 +157,69 @@ 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' + # 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' runs-on: ubuntu-latest @@ -290,7 +372,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/cython.rst b/docs/developerguide/cython.rst new file mode 100644 index 000000000..7facfde9f --- /dev/null +++ b/docs/developerguide/cython.rst @@ -0,0 +1,244 @@ +.. _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-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-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 +========================= + +**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``'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*: + + .. 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. + +* ``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. + +.. _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 new file mode 100644 index 000000000..f6236c29a --- /dev/null +++ b/docs/developerguide/free_threading.rst @@ -0,0 +1,230 @@ +.. _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: + +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 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 + + 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 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. + +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: + +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``. + +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 e5d3bbf0f..292bdae02 100644 --- a/docs/developerguide/index.rst +++ b/docs/developerguide/index.rst @@ -12,4 +12,6 @@ overview partition_assignor + cython + free_threading diff --git a/docs/includes/settingref.txt b/docs/includes/settingref.txt index 669894cff..c1e4e104d 100644 --- a/docs/includes/settingref.txt +++ b/docs/includes/settingref.txt @@ -282,6 +282,61 @@ 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.14.0 + +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) + +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 + 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 42e06bc28..f763164e6 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 @@ -6,6 +7,8 @@ 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 cdef class StreamIterator: @@ -34,6 +37,7 @@ cdef class StreamIterator: object topics object acks_enabled_for object _skipped_value + bint cython_optimizations def __init__(self, object stream): self.stream = stream @@ -52,6 +56,11 @@ 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. + # 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 @@ -111,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( @@ -188,11 +206,35 @@ 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. + # + # 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 self.chan_queue_empty: + if not self.cython_optimizations: + return (True, None) + 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/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..cad2c3871 100644 --- a/faust/transport/_cython/conductor.pyx +++ b/faust/transport/_cython/conductor.pyx @@ -1,7 +1,9 @@ # cython: language_level=3 +# cython: freethreading_compatible=True 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: @@ -19,6 +21,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): @@ -32,6 +35,11 @@ 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. 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. @@ -68,10 +76,35 @@ 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 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 delivered.add(chan) if full: @@ -92,7 +125,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) @@ -108,13 +146,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/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/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/settings/settings.py b/faust/types/settings/settings.py index 8f68b2cf4..a30d7a4c1 100644 --- a/faust/types/settings/settings.py +++ b/faust/types/settings/settings.py @@ -604,6 +604,59 @@ def agent_supervisor(self) -> Type[SupervisorStrategyT]: restarted). """ + @sections.Common.setting( + params.Bool, + version_introduced="0.14.0", + 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) + + 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 + two faults were and how the implementations are held level. + """ + @sections.Common.setting( params.Seconds, env_name="BLOCKING_TIMEOUT", 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/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/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/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/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/requirements/test.txt b/requirements/test.txt index f5fd3e24d..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 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" + ) diff --git a/tests/unit/test_cython_parity.py b/tests/unit/test_cython_parity.py new file mode 100644 index 000000000..86b5d6b39 --- /dev/null +++ b/tests/unit/test_cython_parity.py @@ -0,0 +1,248 @@ +"""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 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. + +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 +@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. + + ``_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 +@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. + + 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 + + +@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/test_free_threading.py b/tests/unit/test_free_threading.py new file mode 100644 index 000000000..f65d9943d --- /dev/null +++ b/tests/unit/test_free_threading.py @@ -0,0 +1,158 @@ +"""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 importlib.util +import sys +import sysconfig + +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() + + +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 _free_threaded_build(), + reason="not a free-threaded (PEP 703) build", +) + + +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 +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: + """Importing a faust extension must leave the GIL disabled.""" + 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]) + + 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}" + ) diff --git a/tests/unit/transport/test_conductor_parity.py b/tests/unit/transport/test_conductor_parity.py new file mode 100644 index 000000000..54c3ef060 --- /dev/null +++ b/tests/unit/transport/test_conductor_parity.py @@ -0,0 +1,559 @@ +"""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``. + +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 +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.sensors import Monitor +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), + # 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), + "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.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.""" + + 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 +@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() + + 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.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.""" + + 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.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. + + 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.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. + + 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.conf(cython_optimizations=True) +@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.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. + + 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(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``. + + 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): + 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": 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) + 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 +@pytest.mark.asyncio +@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. + + ``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) + + +@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 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." + )