Free-threading support, and repairs to the Cython accelerators behind an opt-in

.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

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={'<ver>': '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_<name>`` or ``_Py<Name>`` 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.