Free-threading support, and repairs to the Cython accelerators behind an opt-in - #762
Open
wbarnha wants to merge 6 commits into
Open
Free-threading support, and repairs to the Cython accelerators behind an opt-in#762wbarnha wants to merge 6 commits into
wbarnha wants to merge 6 commits into
Conversation
A free-threaded interpreter re-enables the GIL for the whole process the
moment it imports an extension module that has not declared
`Py_mod_gil = Py_MOD_GIL_NOT_USED`, and reports it only through a
RuntimeWarning. All three of faust's Cython extensions were in that
state, so importing faust on 3.13t/3.14t silently turned free-threading
off:
RuntimeWarning: The global interpreter lock (GIL) has been enabled to
load module 'faust._cython.windows', which has not declared that it
can run safely without the GIL.
Set `freethreading_compatible=True` in the three .pyx files, which is
what makes Cython emit the slot. The modules qualify: windows.pyx holds
cdef doubles written once in __init__ and read-only after, and
streams.pyx / conductor.pyx hold per-instance Python references with all
shared state in ordinary Python containers.
Two things made the loss easy to reintroduce invisibly, so both are
pinned down:
* The directive only exists in Cython 3.1+, and older Cython ignores
unknown directives rather than failing -- a 3.0 build would emit no
declaration and no diagnostic. Add a `cython>=3.1` floor for 3.13+
in build-system.requires, and pin cibuildwheel's before-build the
same way.
* Nothing fails when the declaration is missing. Add
tests/unit/test_free_threading.py, which imports each extension in a
subprocess and asserts the GIL is still off. The subprocess drops
PYTHON_GIL from its environment, or the CI job's PYTHON_GIL=0 would
make the assertion vacuous. It skips on a GIL interpreter.
With that, the full unit + functional suite passes on both 3.13t and
3.14t against the compiled extensions with the GIL genuinely disabled
(2211 passed), so drop the `cp31?t-*` cibuildwheel skip the previous
comment said to drop "once faust is verified free-threading-safe", and
add `enable = ["cpython-freethreading"]` so cp313t is built alongside
cp314t.
The new `free-threaded` job covers both interpreters and gates merges,
since it is what verifies the wheels being published. Two things it
does differently from the other test jobs, both necessary:
* It installs requirements/freethreading.txt, not test.txt. Parts of
test.txt cannot be built on a free-threaded interpreter at all --
twine pulls in cffi, which refuses to build on 3.13t, and hypothesis
6.130+ ships a PyO3 extension that does not support 3.13t either.
The new file documents every omission and pin.
* It builds the extensions in place. pytest runs from the repo root,
so `import faust` resolves to the source tree, and the accelerated
implementations are imported behind `try: ... except ImportError`.
Without a .so next to the .pyx the fallback engages silently and the
run exercises pure Python regardless of USE_CYTHON -- which is also
true of the existing USE_CYTHON=true matrix legs.
Document the above in docs/developerguide/free_threading.rst, along with
two findings that are not fixed here: aiokafka's extensions have not made
the declaration either, so a real worker gets the GIL back when the
transport driver loads; and Message.ack/decref is a non-atomic
read-modify-write that loses final acks under real parallelism. The
latter is not reachable from faust's own code, which acks from the event
loop, but is reachable via the public Event.ack.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01K8qT5E3rnSXvw7ibNLXrVr
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #762 +/- ##
==========================================
- Coverage 96.06% 96.05% -0.01%
==========================================
Files 103 104 +1
Lines 11072 11086 +14
Branches 1191 1189 -2
==========================================
+ Hits 10636 10649 +13
Misses 345 345
- Partials 91 92 +1 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
The optional Cython accelerators were never executed by a single test. `pip install .` compiles them into site-packages, but pytest runs from the repository root, so `import faust` resolves to the source tree and every accelerated import sits behind `try: ... except ImportError`. With no .so next to the .pyx the fallback engaged silently, so the `use-cython: true` matrix legs differed from the `false` ones only in whether the build step succeeded. Build the extensions in place on those legs, and add FAUST_REQUIRE_CYTHON, which turns the silent fallback into a failure so the gap cannot quietly reopen. This matters beyond this branch: the parity tests proposed in #751 note they otherwise "just run the pure-Python one twice", which in CI was always. ## The bug this uncovered `StreamIterator._try_get_quick_value` carried two faults that concealed each other. `chan_queue_empty` holds the bound `queue.empty` method: # streams.py # streams.pyx if chan_queue_empty(): if self.chan_queue_empty: A bound method is always truthy, so the extension always reported "queue empty" and took the awaiting path. That made the `else` unreachable -- which hid the fact that it returned the bare value from `get_nowait()` instead of the `(need_slow_get, value)` pair the caller unpacks. Had the fast path ever run, `next()` would have raised TypeError, or silently mis-unpacked a two-element value into `need_slow_get, channel_value`. Both are fixed together; fixing only the condition would have activated the broken return. The pure-Python twin has always had this right, so this restores the fast path the extension was meant to provide and brings the two implementations back into agreement. Net effect: the compiled iterator has been doing strictly more work than the pure Python it was meant to accelerate, for as long as it has existed. ## Tests tests/unit/test_cython_parity.py covers the guard, window parity (HoppingWindow/SlidingWindow against their _Py twins across step boundaries), and both branches of the queue fast path. The stream tests drive `StreamIterator.next()` directly rather than `async for`, which would need a running worker, and count calls to `Channel.__anext__` -- the awaiting path -- because that is the only clean signal. The two obvious alternatives both fail: `get_nowait` is called by `Queue.get` on the slow path too, and `empty` is called from inside `get_nowait`, so both fire either way and only the counts differ. Verified in both directions: reintroducing the bug fails the test with 5 `__anext__` calls for 5 already-queued values, against 0 when fixed. Suite passes in every configuration: extensions built (2254 passed), absent (2207 passed, parity tests skipped), and free-threaded 3.14t with PYTHON_GIL=0 (2258 passed). ## Docs docs/developerguide/cython.rst records how to test the compiled code, the drift history that motivates parity tests (#608, the on_topic_buffer_full defect left unfixed because fixing one twin alone would desynchronise them, and the fast-path pair above), and the conventions for adding an accelerator -- including that the wins concentrate in per-call arithmetic, not in code whose body is mostly `await`. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01K8qT5E3rnSXvw7ibNLXrVr
The topic conductor is the per-message inner loop of a worker, and it exists twice: `ConductorHandler` in the extension, and the `on_message` closure from `ConductorCompiler.build`. Neither was covered -- the existing conductor tests replace the handler with an AsyncMock and assert it was called, so the fan-out, event reuse, buffer-pressure callbacks, full-queue path and decode-error propagation were untested on both sides. Both handlers take `(conductor, tp, channels)` and are awaited with a Message, so they can be driven over the same input and compared. tests/unit/transport/test_conductor_parity.py does that for each of those paths and diffs a full record of the outcome: which events reached which channels, refcount and acked state, decode counts, and every sensor and consumer callback. Both implementations run against the *same* conductor and the same `channels` set, one after the other, rather than two separately-built environments. `channels` is a set of Topic objects hashed by identity, so two separate sets iterate in unrelated orders, and anything order-sensitive -- which channel decodes first, which ones a mid-fan-out decode error reaches -- would differ for reasons unrelated to the implementations. Sharing the set removes that variable; `reset()` clears queues and recorded callbacks between runs. ## What it found `ConductorHandler` never reused a decoded event. The conductor is supposed to deserialize once and reuse it for every channel whose `(key_type, value_type)` matches, but `event_keyid` was only ever assigned from `_decode()`, which returned it *unchanged* on the first pass. It stayed None forever, the reuse branch was dead, and every subscribed channel re-deserialized the payload. That masked a second fault. Had the keyid ever been set, a mismatched pair fell off the end of `_decode` and returned a bare None, which unpacking into two names raises TypeError on. Fixing the reuse alone would have turned a silent inefficiency into a crash on any topic whose subscribers declare different key or value types -- confirmed by building that partial fix and watching the new heterogeneous-keyid test fail with `TypeError: 'NoneType' object is not iterable` at the unpack. This is the same double-bug shape as `_try_get_quick_value` in streams.pyx, arrived at independently: a dead optimization whose deadness concealed that it was also wrong. It was not only a performance difference. A channel whose event is reused never calls `decode`, so a channel that would have failed to deserialize raised no error under the pure-Python conductor and raised one under the extension -- changing which channels received the message and how many acks it got. The fix ports conductor.py's loop faithfully: `event`/`event_keyid` stay pinned to the first channel, and a channel with a different pair gets its own `dest_event` without displacing the pinned one. `_decode` is gone; `keyid` and `dest_event` were already declared in `__call__` and unused, which suggests this is what it was meant to be. ## Verification 16 parity tests, covering fan-out over 1/2/3 channels, no subscribers, batches, event reuse for matching keyids, per-channel decode for differing keyids, decode errors (whole fan-out and single channel), the full-queue path and the pressure callbacks. Before the fix 6 of them failed, all tracing to that one root cause; after it, all pass. Full suite green in every configuration: extensions built (2270 passed), absent (2207 passed, parity skipped), NO_CYTHON=1, and free-threaded 3.14t under PYTHON_GIL=0 (2274 passed). Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01K8qT5E3rnSXvw7ibNLXrVr
`Monitor.topic_buffer_full` is a `Counter[TP]`, and two paths report into it: the pressure-high callback, which passes a TP, and the full-queue path, which passed the *channel*. The same partition therefore accumulated under two different keys depending on which path noticed the buffer was full -- splitting its count, and adding a second `/stats` entry labelled by channel for a partition already listed by TP. Both implementations had it, which is why it went unfixed for so long: the comment in faust/transport/conductor.py recorded the defect and explicitly declined to fix it, because correcting one twin alone would have made the two disagree. With the parity suite in place that objection is gone -- both are corrected here, together, and the suite holds them level. The `# type: ignore[arg-type]` on the call goes away with it; `mypy -p faust` is clean without it, which is the type checker confirming the argument is now the one the sensor declares. ## Note on what parity testing does not do The conductor parity tests were green throughout, before and after. Both implementations passed the channel, so they agreed with each other perfectly while both were wrong. A differential test only finds *divergence*; a shared mistake is invisible to it. So the coverage added here is deliberately not another comparison: * the full-queue parity test now records the sensor's *argument* rather than a call count, and asserts it equals the TP; * a new test drives a real `Monitor` through the full-queue path and asserts every key of `topic_buffer_full` is a TP. It is parametrised over both implementations rather than comparing them, and runs against the pure-Python conductor even when the extension is absent, since the defect was in both. Verified by reverting both twins and confirming each new assertion fails: `Got: [<Topic: foo0@...>]` and `keyed it by ['Topic']`. Suite green in every configuration: extensions built (2272 passed), absent (2208 passed), free-threaded 3.14t under PYTHON_GIL=0 (2276 passed), and `mypy -p faust` clean. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01K8qT5E3rnSXvw7ibNLXrVr
Two of the Cython fast paths never ran, each guarded by a condition that
could not become true, so the extensions quietly did more work than the
Python they were meant to accelerate. Repairing them (in the two PRs
below this one) activates code that has by definition never executed in
production. Gate it.
`cython_optimizations` defaults to False. With it off the extensions
behave exactly as the released versions do, so upgrading changes
nothing; users opt in per app:
app = faust.App('myapp', cython_optimizations=True)
or `CYTHON_OPTIMIZATIONS=1` in the environment (prefixed when
`env_prefix` is set, like every other env-backed setting).
The flag is read once per StreamIterator and once per ConductorHandler
-- so once per stream and once per assigned TP, not per message -- into
a `bint`, leaving a predictable branch on the hot path rather than an
attribute lookup into `app.conf`.
## What it gates, and what it does not
Gated:
* `StreamIterator._try_get_quick_value` -- taking values already in
the channel queue instead of always awaiting.
* `ConductorHandler` event reuse -- decoding once and reusing the
event across channels with matching key/value types, instead of
deserializing once per subscribed channel.
Not gated: the `on_topic_buffer_full` argument fix. That one was wrong
in *both* implementations, is not Cython-specific, and produced a metric
that was simply incorrect -- gating a wrong metric key behind a
"Cython improvements" flag would be incoherent. It applies always.
## Consequence worth stating plainly
While the setting is off, the Cython and pure-Python paths genuinely
differ. That is not new -- it is what has shipped for years -- and the
flag does not introduce the divergence, only makes it selectable. The
sharpest case is the conductor: a reused event is never decoded again,
so a channel whose payload would fail to deserialize raises no error
when the event is reused and raises one when it is not, changing which
channels receive a message and how many acks it takes.
So the parity suites now run with the setting on, which is the
configuration in which the two implementations are supposed to agree.
Each suite also gains a test pinning the default-off behaviour, so the
historical path -- the one most users will actually run -- stays
covered: 5 awaits for 5 queued values in the iterator, one decode per
channel in the conductor.
## Verification
Suite green in every configuration: extensions built (2274 passed),
absent (2208 passed), free-threaded 3.14t under PYTHON_GIL=0 (2278
passed). `mypy -p faust` clean, `extra/tools/verify_doc_defaults.py`
clean, docs build clean with the setting rendered into the
configuration reference.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01K8qT5E3rnSXvw7ibNLXrVr
The setting is transitional -- it exists so the repaired Cython fast paths are adopted deliberately rather than arriving in an upgrade, and it is meant to be removed, not kept. Retiring it naively has a trap in it, which this closes before anyone walks into it. `Param.__get__` emits a UserWarning on *every read* of a setting once `version_deprecated` is set, and faust reads this one itself: once per Stream, once per assigned partition. Setting `version_deprecated` would therefore make faust warn at itself, at a rate that scales with the deployment, about a setting the user most likely never set and cannot act on. Measured before the change: three StreamIterator constructions, three warnings. Both extensions now read the flag through `faust.utils.optin.cython_optimizations_enabled`, which takes the value the descriptor stores instead of going through the descriptor. Internal reads stay silent; `app.conf.cython_optimizations` still warns, which is the entire point of deprecating a setting -- a helper that disarmed that too would be worse than the noise, because nobody would ever be told to stop using it. The storage attribute is looked up through the settings registry rather than hard-coded, so renaming the setting cannot silently turn this into a read of a missing attribute. Deliberately not `warnings.catch_warnings()`: it manipulates global state and is not thread-safe, which matters on the free-threaded builds this branch series added support for. ## Tests tests/unit/utils/test_optin.py pins both halves of the contract -- the internal read silent under deprecation, the public read still warning -- plus an end-to-end check that three stream iterators and three conductor handlers produce zero warnings with the setting marked deprecated (three and three before). The deprecation is applied by a fixture that restores the param afterwards, so the tests need no released deprecation to run. ## Docs The developer guide gains the intended sequence: ships off, default flipped once there is real-world evidence (parity passing is necessary but not sufficient -- it only proves the two implementations agree under test), deprecated, then removed along with the branches, the helper and the default-off tests. The setting's own docstring says it is transitional, so it does not read as permanent API. Suite green in every configuration: extensions built (2280 passed), absent (2213 passed), free-threaded 3.14t under PYTHON_GIL=0 (2284 passed). mypy, verify_doc_defaults and the docs build all clean. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01K8qT5E3rnSXvw7ibNLXrVr
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Description
Consolidates what was #762–#765 into a single branch. Six commits, each self-contained; the story runs in order.
It started as "investigate free-threading support", and the first fix uncovered that the Cython accelerators were never executed by a single test — which turned out to be hiding two live bugs in them.
1. Free-threaded CPython (PEP 703)
A free-threaded interpreter re-enables the GIL for the whole process the moment it imports an extension that hasn't declared
Py_mod_gil, reporting it only through aRuntimeWarning. All three of faust's extensions were in that state, so importing faust on 3.13t/3.14t silently turned free-threading off.The three
.pyxfiles now setfreethreading_compatible=True. Two things made the loss easy to reintroduce invisibly, so both are pinned: acython>=3.1floor for 3.13+ (older Cython ignores unknown directives rather than failing), andtests/unit/test_free_threading.py, which imports each extension in a subprocess and asserts the GIL is still off.The full suite passes on both 3.13t and 3.14t against the compiled extensions with the GIL genuinely disabled, so the
cp31?t-*cibuildwheel skip is dropped and a gatingfree-threadedCI job covers both.Known limitation, not fixable here:
aiokafka's extensions haven't made the declaration either, so a real worker gets the GIL back when the transport driver loads. Upstream.2. The compiled code was never tested
pip install .compiles the extensions into site-packages, but pytest runs from the repository root — soimport faustresolves to the source tree, and every accelerated import sits behindtry: ... except ImportError. The fallback engaged silently. Theuse-cython: truematrix legs differed from thefalseones only in whether the build step succeeded.Those legs now build in place, and
FAUST_REQUIRE_CYTHONturns the silent fallback into a failure. This matters beyond this branch: the parity tests proposed in #751 note they otherwise "just run the pure-Python one twice" — which in CI was always.3. Two bugs that were hiding there
Both are the same shape, found independently: a dead optimization whose deadness concealed that it was also wrong.
StreamIterator._try_get_quick_value—chan_queue_emptyholds the boundqueue.emptymethod:Always truthy → always the awaiting path → the
elseunreachable, which hid that it returned the bareget_nowait()value instead of the(need_slow_get, value)pair the caller unpacks.ConductorHandlerevent reuse —event_keyidwas only ever assigned from_decode(), which returned it unchanged on the first pass. It stayedNoneforever, so every subscribed channel re-deserialized the payload. Had the keyid ever been set, a mismatched pair fell off the end of_decodereturning a bareNone; unpacking that raisesTypeError. Confirmed by building the partial fix and watching the new heterogeneous-keyid test fail exactly that way.Not only performance: a reused event is never decoded again, so a channel whose payload would fail to deserialize raised no error under pure Python and raised one under the extension — changing which channels received a message and how many acks it took.
4.
topic_buffer_fullkeyed by TPMonitor.topic_buffer_fullis aCounter[TP], but the full-queue path passed the channel, so the same partition accumulated under two keys. Both twins had it, which is whyconductor.pyrecorded it as deliberately unfixed — correcting one alone would have made them disagree. With the parity suite in place, both are corrected together.Worth noting what did not catch it: the parity tests were green throughout, because both implementations were wrong identically. A differential test only finds divergence. So the coverage added for this asserts what the value is, not that both sides agree.
5. The opt-in
The repaired fast paths have, by definition, never run in production.
cython_optimizationsgates them and defaults toFalse— upgrading changes nothing:or
CYTHON_OPTIMIZATIONS=1(prefixed whenenv_prefixis set). Read once per stream and per assigned TP into abint, not per message.Not gated: the
topic_buffer_fullfix — wrong in both implementations, not Cython-specific, and gating a wrong metric key behind a "Cython improvements" flag would be incoherent.While the setting is off the two paths genuinely differ. That is not new; the flag makes it selectable rather than introducing it. So the parity suites run with it on, and each suite pins the default-off behaviour separately.
6. Retiring the flag is a two-line change
The setting is transitional. Retiring it has a trap:
Param.__get__warns on every read onceversion_deprecatedis set, and faust reads this one itself — once per stream, once per partition. Deprecating it naively would make faust warn at itself at a rate that scales with the deployment. Measured: threeStreamIteratorconstructions, three warnings.Both extensions now read it through
faust.utils.optin.cython_optimizations_enabled, which takes the stored value. Internal reads stay silent;app.conf.cython_optimizationsstill warns — a helper that disarmed that too would be worse than the noise. Tests pin both halves, and the developer guide records the intended sequence through to removal.Verification
FAUST_REQUIRE_CYTHON=1)NO_CYTHON=1PYTHON_GIL=0mypy -p faustextra/tools/verify_doc_defaults.pyflake8/black/isortclean; docs build clean. Every fix was verified in both directions — reintroduced, watched the test fail, restored.Docs:
docs/developerguide/cython.rst(drift history, how to test the compiled code, the opt-in and its retirement plan) anddocs/developerguide/free_threading.rst.One note: the
settingref.txtentry was added by hand rather than bymake configref— the committed file was generated by different tooling, and regenerating reformats every block in it. Worth a separate PR to resync that generator.Supersedes #763, #764 and #765, now closed.