Add three Cython accelerators: offset commit, record scheduler, sensor fan-out - #751
Open
wbarnha wants to merge 5 commits into
Open
Add three Cython accelerators: offset commit, record scheduler, sensor fan-out#751wbarnha wants to merge 5 commits into
wbarnha wants to merge 5 commits into
Conversation
Faust already ships optional Cython implementations of the window types,
the stream iterator and the topic conductor. This extends that to three
more hot paths, found by profiling the per-message and per-commit work.
Each follows the existing pattern: the pure-Python implementation stays
and is used whenever the extension could not be built or NO_CYTHON is
set, so this is purely additive.
faust/utils/_cython/functional.pyx
first_consecutive_run(), a dedicated helper for the one thing
Consumer._new_offset actually needs from consecutive_numbers(): the
first run. The groupby() version builds a tuple, calls a Python key
function and creates a group generator per acked offset, and runs
once per assigned partition on every commit. Because it blocks the
event loop, the win shows up as latency: for 100k un-committed
offsets, 2.71ms -> 0.55ms.
faust/transport/_cython/scheduler.pyx
A C-level cursor replacing the round-robin generator returned by
DefaultSchedulingStrategy.records_iterator, which also takes over
TopicBuffer's inner per-partition generator. This runs once per
record fetched from the broker: ~300-490ns -> ~50-70ns.
faust/sensors/_cython/base.pyx
A cdef base class carrying the four sensor hooks that fire on every
message; SensorDelegate subclasses it and keeps its ~20 occasional
hooks in plain Python. on_stream_event_in, which builds a dict per
event, goes 374ns -> 182ns.
Behaviour is preserved rather than approximated. Both scheduler
implementations re-read the topic index and each topic's buffer map at
every pass and pop drained entries from them, so mutating either
mid-iteration gives the same result; a TopicBuffer subclass is still
driven through next(); map_from_records and records_iterator remain
overridable. first_consecutive_run stops consuming as soon as the run
ends, so a shared iterator is left where the pure-Python version leaves
it, and it re-reads the list length while scanning because a custom
__sub__ can resize the list underneath it. The sensor set is read live
rather than snapshotted, so direct mutation behaves identically.
Tests run every new code path against both implementations, including a
randomised differential test over 200 scheduler topologies. The full
suite passes with and without NO_CYTHON.
extra/tools/benchmark_cython.py times each accelerator against its
pure-Python counterpart in the same interpreter.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019oX4oGCQGgvPJHabjwfaBk
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #751 +/- ##
=======================================
Coverage 96.06% 96.06%
=======================================
Files 103 103
Lines 11072 11094 +22
Branches 1191 1193 +2
=======================================
+ Hits 10636 10658 +22
Misses 345 345
Partials 91 91 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
The copartitioned-assignor property tests run on PyPy since the PyPy skips were removed. They carried a fixed 4000ms Hypothesis deadline, but PyPy's JIT warmup makes a single example's timing vary by over a second, so runs intermittently exceed it and Hypothesis reports DeadlineExceeded / FlakyFailure -- failing the (non-required) PyPy leg and turning master CI red even though the assignment produced is valid. These tests assert assignment correctness, not performance, so set deadline=None (as the DeadlineExceeded message itself recommends). The property assertions are unchanged and the job timeout still guards against a real hang. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL
Member
Author
|
The The failure was Evidence it isn't caused by this branch:
Hypothesis shrinks to the same slow example every run, so it was not going to clear on its own. Everything else is green, and Generated by Claude Code |
Follows the "faster code via static typing" guidance: declare return types on cdef functions, and type parameters and locals rather than leaving them implicit. The substantive change is on the two _start_pass methods, which had no declared return type. They are now `cdef int ... except -1`. Both call into Python (dict.pop, .items(), the _TopicCursor constructor) and so can raise, and the except clause states that rather than depending on a compiler default -- Cython 3 propagates from a bare cdef, but Cython 0.x swallowed unless an except clause was given, and legacy_implicit_noexcept restores that behaviour. Verified both ways: an exception raised inside _start_pass now propagates identically through the Cython and pure-Python iterators. Also types the len() results as Py_ssize_t instead of letting them box into Python ints, and types the two sensor-delegate parameters that were bare. Note this is a correctness and clarity change, not a speed one: the scheduler benchmark is unmoved at 5.8-7.7x over pure Python, because the loops were already fully typed. Parameters that face callers are deliberately left as `object` -- typing them to a builtin would add an exact-type runtime check, which is what made an OrderedDict fail against a `cdef dict` local earlier in this branch. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019oX4oGCQGgvPJHabjwfaBk
Reported in review of #751. _start_pass skipped rebuilding its snapshot when the mapping's length was unchanged since the last pass. That check cannot tell "unchanged" from "swapped": replacing index["foo"] with a new TopicBuffer under the same key leaves the length identical, so the shortcut was taken, the stale cursor kept being used, and every record in the replacement was silently dropped. The `cursor.source is not buffer` check that exists to catch exactly this lived inside the rebuild, which the shortcut skipped. Reproduced against the pure-Python iterator, which re-reads the mapping every pass and so gets it right: python: [1, 10, 11] cython: [1, 2] <- 10 and 11 never delivered The same flaw was present one level down in _TopicCursor, where an iterator swapped in under an existing TP left the partition cursor stale. The report only named the topic level; both are fixed. Both shortcuts are replaced by _snapshot_is_current(), which compares the cached snapshot against the live mapping entry by entry, by identity. It walks the mapping but allocates nothing, which is what still makes reusing the snapshot worthwhile. This costs real throughput, and the PR body is updated rather than left overstating it: the scheduler was 5.7-7.7x over pure Python and is now 2.2-2.8x. A pass must consult the live mapping to be correct, and a single-partition topic reaches a pass boundary on every record. An unconditional rebuild was measured too, at 1.6-2.1x, so the identity check is worth keeping over simply rebuilding. Adds regression tests for both levels, run against both implementations. Swapping in a topic under a *different* name is deliberately not covered: the pure-Python generator iterates the live dict, so CPython raises "dictionary keys changed during iteration" -- undefined behaviour in Python itself rather than a guarantee to pin. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019oX4oGCQGgvPJHabjwfaBk
Member
Author
|
Confirmed and fixed in Reproduced before touching anything, against the pure-Python iterator as the reference: The length check could not distinguish "unchanged" from "swapped", so the shortcut was taken and the stale cursor kept draining the old buffer. The The same flaw was present one level down, which the report did not mention: in On the suggested remedies — I took the second one, but not by rebuilding unconditionally. Both shortcuts are now
So the fix costs most of the speedup — the scheduler was 5.7–7.7x over pure Python and is now 2.2–2.8x. A pass has to consult the live mapping to be correct, and a single-partition topic reaches a pass boundary on every record. I have corrected the benchmark table in the PR description rather than leave it overstating the result. Both of your regression scenarios are now tests, parametrised over both implementations, plus the partition-level analogue. The randomised differential test is at 4000/4000. One scenario I deliberately did not pin: replacing a topic under a different name. The pure-Python generator iterates the live dict, so CPython raises Generated by Claude Code |
wbarnha
pushed a commit
that referenced
this pull request
Aug 7, 2026
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
This was referenced Aug 7, 2026
wbarnha
pushed a commit
that referenced
this pull request
Aug 10, 2026
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
wbarnha
added a commit
that referenced
this pull request
Aug 10, 2026
… an opt-in (#762) * Support free-threaded CPython (PEP 703) on 3.13t and 3.14t A free-threaded interpreter re-enables the GIL for the whole process the moment it imports an extension module that has not declared `Py_mod_gil = Py_MOD_GIL_NOT_USED`, and reports it only through a RuntimeWarning. All three of faust's Cython extensions were in that state, so importing faust on 3.13t/3.14t silently turned free-threading off: RuntimeWarning: The global interpreter lock (GIL) has been enabled to load module 'faust._cython.windows', which has not declared that it can run safely without the GIL. Set `freethreading_compatible=True` in the three .pyx files, which is what makes Cython emit the slot. The modules qualify: windows.pyx holds cdef doubles written once in __init__ and read-only after, and streams.pyx / conductor.pyx hold per-instance Python references with all shared state in ordinary Python containers. Two things made the loss easy to reintroduce invisibly, so both are pinned down: * The directive only exists in Cython 3.1+, and older Cython ignores unknown directives rather than failing -- a 3.0 build would emit no declaration and no diagnostic. Add a `cython>=3.1` floor for 3.13+ in build-system.requires, and pin cibuildwheel's before-build the same way. * Nothing fails when the declaration is missing. Add tests/unit/test_free_threading.py, which imports each extension in a subprocess and asserts the GIL is still off. The subprocess drops PYTHON_GIL from its environment, or the CI job's PYTHON_GIL=0 would make the assertion vacuous. It skips on a GIL interpreter. With that, the full unit + functional suite passes on both 3.13t and 3.14t against the compiled extensions with the GIL genuinely disabled (2211 passed), so drop the `cp31?t-*` cibuildwheel skip the previous comment said to drop "once faust is verified free-threading-safe", and add `enable = ["cpython-freethreading"]` so cp313t is built alongside cp314t. The new `free-threaded` job covers both interpreters and gates merges, since it is what verifies the wheels being published. Two things it does differently from the other test jobs, both necessary: * It installs requirements/freethreading.txt, not test.txt. Parts of test.txt cannot be built on a free-threaded interpreter at all -- twine pulls in cffi, which refuses to build on 3.13t, and hypothesis 6.130+ ships a PyO3 extension that does not support 3.13t either. The new file documents every omission and pin. * It builds the extensions in place. pytest runs from the repo root, so `import faust` resolves to the source tree, and the accelerated implementations are imported behind `try: ... except ImportError`. Without a .so next to the .pyx the fallback engages silently and the run exercises pure Python regardless of USE_CYTHON -- which is also true of the existing USE_CYTHON=true matrix legs. Document the above in docs/developerguide/free_threading.rst, along with two findings that are not fixed here: aiokafka's extensions have not made the declaration either, so a real worker gets the GIL back when the transport driver loads; and Message.ack/decref is a non-atomic read-modify-write that loses final acks under real parallelism. The latter is not reachable from faust's own code, which acks from the event loop, but is reachable via the public Event.ack. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01K8qT5E3rnSXvw7ibNLXrVr * Make the Cython path testable, and fix the bug that hid in it The optional Cython accelerators were never executed by a single test. `pip install .` compiles them into site-packages, but pytest runs from the repository root, so `import faust` resolves to the source tree and every accelerated import sits behind `try: ... except ImportError`. With no .so next to the .pyx the fallback engaged silently, so the `use-cython: true` matrix legs differed from the `false` ones only in whether the build step succeeded. Build the extensions in place on those legs, and add FAUST_REQUIRE_CYTHON, which turns the silent fallback into a failure so the gap cannot quietly reopen. This matters beyond this branch: the parity tests proposed in #751 note they otherwise "just run the pure-Python one twice", which in CI was always. ## The bug this uncovered `StreamIterator._try_get_quick_value` carried two faults that concealed each other. `chan_queue_empty` holds the bound `queue.empty` method: # streams.py # streams.pyx if chan_queue_empty(): if self.chan_queue_empty: A bound method is always truthy, so the extension always reported "queue empty" and took the awaiting path. That made the `else` unreachable -- which hid the fact that it returned the bare value from `get_nowait()` instead of the `(need_slow_get, value)` pair the caller unpacks. Had the fast path ever run, `next()` would have raised TypeError, or silently mis-unpacked a two-element value into `need_slow_get, channel_value`. Both are fixed together; fixing only the condition would have activated the broken return. The pure-Python twin has always had this right, so this restores the fast path the extension was meant to provide and brings the two implementations back into agreement. Net effect: the compiled iterator has been doing strictly more work than the pure Python it was meant to accelerate, for as long as it has existed. ## Tests tests/unit/test_cython_parity.py covers the guard, window parity (HoppingWindow/SlidingWindow against their _Py twins across step boundaries), and both branches of the queue fast path. The stream tests drive `StreamIterator.next()` directly rather than `async for`, which would need a running worker, and count calls to `Channel.__anext__` -- the awaiting path -- because that is the only clean signal. The two obvious alternatives both fail: `get_nowait` is called by `Queue.get` on the slow path too, and `empty` is called from inside `get_nowait`, so both fire either way and only the counts differ. Verified in both directions: reintroducing the bug fails the test with 5 `__anext__` calls for 5 already-queued values, against 0 when fixed. Suite passes in every configuration: extensions built (2254 passed), absent (2207 passed, parity tests skipped), and free-threaded 3.14t with PYTHON_GIL=0 (2258 passed). ## Docs docs/developerguide/cython.rst records how to test the compiled code, the drift history that motivates parity tests (#608, the on_topic_buffer_full defect left unfixed because fixing one twin alone would desynchronise them, and the fast-path pair above), and the conventions for adding an accelerator -- including that the wins concentrate in per-call arithmetic, not in code whose body is mostly `await`. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01K8qT5E3rnSXvw7ibNLXrVr * Add end-to-end conductor parity tests, and fix the divergence they found The topic conductor is the per-message inner loop of a worker, and it exists twice: `ConductorHandler` in the extension, and the `on_message` closure from `ConductorCompiler.build`. Neither was covered -- the existing conductor tests replace the handler with an AsyncMock and assert it was called, so the fan-out, event reuse, buffer-pressure callbacks, full-queue path and decode-error propagation were untested on both sides. Both handlers take `(conductor, tp, channels)` and are awaited with a Message, so they can be driven over the same input and compared. tests/unit/transport/test_conductor_parity.py does that for each of those paths and diffs a full record of the outcome: which events reached which channels, refcount and acked state, decode counts, and every sensor and consumer callback. Both implementations run against the *same* conductor and the same `channels` set, one after the other, rather than two separately-built environments. `channels` is a set of Topic objects hashed by identity, so two separate sets iterate in unrelated orders, and anything order-sensitive -- which channel decodes first, which ones a mid-fan-out decode error reaches -- would differ for reasons unrelated to the implementations. Sharing the set removes that variable; `reset()` clears queues and recorded callbacks between runs. ## What it found `ConductorHandler` never reused a decoded event. The conductor is supposed to deserialize once and reuse it for every channel whose `(key_type, value_type)` matches, but `event_keyid` was only ever assigned from `_decode()`, which returned it *unchanged* on the first pass. It stayed None forever, the reuse branch was dead, and every subscribed channel re-deserialized the payload. That masked a second fault. Had the keyid ever been set, a mismatched pair fell off the end of `_decode` and returned a bare None, which unpacking into two names raises TypeError on. Fixing the reuse alone would have turned a silent inefficiency into a crash on any topic whose subscribers declare different key or value types -- confirmed by building that partial fix and watching the new heterogeneous-keyid test fail with `TypeError: 'NoneType' object is not iterable` at the unpack. This is the same double-bug shape as `_try_get_quick_value` in streams.pyx, arrived at independently: a dead optimization whose deadness concealed that it was also wrong. It was not only a performance difference. A channel whose event is reused never calls `decode`, so a channel that would have failed to deserialize raised no error under the pure-Python conductor and raised one under the extension -- changing which channels received the message and how many acks it got. The fix ports conductor.py's loop faithfully: `event`/`event_keyid` stay pinned to the first channel, and a channel with a different pair gets its own `dest_event` without displacing the pinned one. `_decode` is gone; `keyid` and `dest_event` were already declared in `__call__` and unused, which suggests this is what it was meant to be. ## Verification 16 parity tests, covering fan-out over 1/2/3 channels, no subscribers, batches, event reuse for matching keyids, per-channel decode for differing keyids, decode errors (whole fan-out and single channel), the full-queue path and the pressure callbacks. Before the fix 6 of them failed, all tracing to that one root cause; after it, all pass. Full suite green in every configuration: extensions built (2270 passed), absent (2207 passed, parity skipped), NO_CYTHON=1, and free-threaded 3.14t under PYTHON_GIL=0 (2274 passed). Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01K8qT5E3rnSXvw7ibNLXrVr * Key topic_buffer_full by TP in both conductors `Monitor.topic_buffer_full` is a `Counter[TP]`, and two paths report into it: the pressure-high callback, which passes a TP, and the full-queue path, which passed the *channel*. The same partition therefore accumulated under two different keys depending on which path noticed the buffer was full -- splitting its count, and adding a second `/stats` entry labelled by channel for a partition already listed by TP. Both implementations had it, which is why it went unfixed for so long: the comment in faust/transport/conductor.py recorded the defect and explicitly declined to fix it, because correcting one twin alone would have made the two disagree. With the parity suite in place that objection is gone -- both are corrected here, together, and the suite holds them level. The `# type: ignore[arg-type]` on the call goes away with it; `mypy -p faust` is clean without it, which is the type checker confirming the argument is now the one the sensor declares. ## Note on what parity testing does not do The conductor parity tests were green throughout, before and after. Both implementations passed the channel, so they agreed with each other perfectly while both were wrong. A differential test only finds *divergence*; a shared mistake is invisible to it. So the coverage added here is deliberately not another comparison: * the full-queue parity test now records the sensor's *argument* rather than a call count, and asserts it equals the TP; * a new test drives a real `Monitor` through the full-queue path and asserts every key of `topic_buffer_full` is a TP. It is parametrised over both implementations rather than comparing them, and runs against the pure-Python conductor even when the extension is absent, since the defect was in both. Verified by reverting both twins and confirming each new assertion fails: `Got: [<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 * Put the repaired Cython fast paths behind an opt-in setting Two of the Cython fast paths never ran, each guarded by a condition that could not become true, so the extensions quietly did more work than the Python they were meant to accelerate. Repairing them (in the two PRs below this one) activates code that has by definition never executed in production. Gate it. `cython_optimizations` defaults to False. With it off the extensions behave exactly as the released versions do, so upgrading changes nothing; users opt in per app: app = faust.App('myapp', cython_optimizations=True) or `CYTHON_OPTIMIZATIONS=1` in the environment (prefixed when `env_prefix` is set, like every other env-backed setting). The flag is read once per StreamIterator and once per ConductorHandler -- so once per stream and once per assigned TP, not per message -- into a `bint`, leaving a predictable branch on the hot path rather than an attribute lookup into `app.conf`. ## What it gates, and what it does not Gated: * `StreamIterator._try_get_quick_value` -- taking values already in the channel queue instead of always awaiting. * `ConductorHandler` event reuse -- decoding once and reusing the event across channels with matching key/value types, instead of deserializing once per subscribed channel. Not gated: the `on_topic_buffer_full` argument fix. That one was wrong in *both* implementations, is not Cython-specific, and produced a metric that was simply incorrect -- gating a wrong metric key behind a "Cython improvements" flag would be incoherent. It applies always. ## Consequence worth stating plainly While the setting is off, the Cython and pure-Python paths genuinely differ. That is not new -- it is what has shipped for years -- and the flag does not introduce the divergence, only makes it selectable. The sharpest case is the conductor: a reused event is never decoded again, so a channel whose payload would fail to deserialize raises no error when the event is reused and raises one when it is not, changing which channels receive a message and how many acks it takes. So the parity suites now run with the setting on, which is the configuration in which the two implementations are supposed to agree. Each suite also gains a test pinning the default-off behaviour, so the historical path -- the one most users will actually run -- stays covered: 5 awaits for 5 queued values in the iterator, one decode per channel in the conductor. ## Verification Suite green in every configuration: extensions built (2274 passed), absent (2208 passed), free-threaded 3.14t under PYTHON_GIL=0 (2278 passed). `mypy -p faust` clean, `extra/tools/verify_doc_defaults.py` clean, docs build clean with the setting rendered into the configuration reference. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01K8qT5E3rnSXvw7ibNLXrVr * Make retiring cython_optimizations a two-line change The setting is transitional -- it exists so the repaired Cython fast paths are adopted deliberately rather than arriving in an upgrade, and it is meant to be removed, not kept. Retiring it naively has a trap in it, which this closes before anyone walks into it. `Param.__get__` emits a UserWarning on *every read* of a setting once `version_deprecated` is set, and faust reads this one itself: once per Stream, once per assigned partition. Setting `version_deprecated` would therefore make faust warn at itself, at a rate that scales with the deployment, about a setting the user most likely never set and cannot act on. Measured before the change: three StreamIterator constructions, three warnings. Both extensions now read the flag through `faust.utils.optin.cython_optimizations_enabled`, which takes the value the descriptor stores instead of going through the descriptor. Internal reads stay silent; `app.conf.cython_optimizations` still warns, which is the entire point of deprecating a setting -- a helper that disarmed that too would be worse than the noise, because nobody would ever be told to stop using it. The storage attribute is looked up through the settings registry rather than hard-coded, so renaming the setting cannot silently turn this into a read of a missing attribute. Deliberately not `warnings.catch_warnings()`: it manipulates global state and is not thread-safe, which matters on the free-threaded builds this branch series added support for. ## Tests tests/unit/utils/test_optin.py pins both halves of the contract -- the internal read silent under deprecation, the public read still warning -- plus an end-to-end check that three stream iterators and three conductor handlers produce zero warnings with the setting marked deprecated (three and three before). The deprecation is applied by a fixture that restores the param afterwards, so the tests need no released deprecation to run. ## Docs The developer guide gains the intended sequence: ships off, default flipped once there is real-world evidence (parity passing is necessary but not sufficient -- it only proves the two implementations agree under test), deprecated, then removed along with the branches, the helper and the default-off tests. The setting's own docstring says it is transitional, so it does not read as permanent API. Suite green in every configuration: extensions built (2280 passed), absent (2213 passed), free-threaded 3.14t under PYTHON_GIL=0 (2284 passed). mypy, verify_doc_defaults and the docs build all clean. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01K8qT5E3rnSXvw7ibNLXrVr * Use importlib.util.find_spec instead of pytest.importorskip in test_free_threading.py; add pytest-run-parallel dependency Co-authored-by: wbarnha <25623043+wbarnha@users.noreply.github.com> * Restore the build toolchain in test.txt, and mirror the Cython 3.13 floor The `use-cython: true` legs build the extensions in place with a direct `python setup.py build_ext --inplace`, which runs against the ambient interpreter rather than pip's isolated build environment. Since 3.12 dropped the ensurepip setuptools seed, that interpreter has no setuptools unless something installs it -- which is what `-r build.txt` in test.txt is for. That line was overwritten by `pytest-run-parallel>=0.10.0`, leaving the comment describing it stranded above the replacement, so 3.12, 3.13 and 3.14 all died at setup.py's first import: ModuleNotFoundError: No module named 'setuptools' 3.10 and 3.11 stayed green only because their runner images still ship setuptools, which is exactly the difference build.txt exists to erase. Restore the include and keep pytest-run-parallel beside the other pytest plugins. build.txt is meant to stay in step with `[build-system].requires`, so it also picks up the `cython>=3.1` floor for 3.13+ added there: older Cython ignores `freethreading_compatible` instead of failing, and now that this file provisions the toolchain for the in-place build, a lower version resolved here would silently produce extensions that re-enable the GIL. Verified locally: `USE_CYTHON=1 python setup.py build_ext --inplace` succeeds and `FAUST_REQUIRE_CYTHON=1 pytest tests/unit tests/functional` runs 2280 passed, 8 skipped against the compiled extensions. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HHM5xcCjiuyh3rPWFzEoP9 * Update version introduced for cython_optimizations * Follow the cython_optimizations version bump into the settings reference 4e363db moved `version_introduced` to 0.14.0 in settings.py but left the configuration reference saying 0.12.2, so the published docs would advertise a version the setting did not ship in. Nothing in CI catches this: `extra/tools/verify_doc_defaults.py` compares defaults, not versions, and this entry is maintained by hand rather than generated -- `make configref` reformats every block in the committed file, which is why it was hand-written in the first place. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HHM5xcCjiuyh3rPWFzEoP9 * Stop the free-threading check from skipping itself when the GIL comes back `requires_free_threading` skipped unless the GIL was *currently* disabled, which is a property the tests themselves can destroy. Any earlier import -- a dependency's extension, a pytest plugin -- re-enables the GIL for the whole process, so the sequence was: GIL re-enabled during collection -> _gil_disabled() false -> all four checks skipped -> "Verify the extensions did not silently re-enable the GIL" exits 0 The step went green in precisely the situation it exists to catch. Reproduced on 3.13t by forcing the condition with PYTHON_GIL=1: 4 skipped, exit 0. Gate on `sysconfig.get_config_var("Py_GIL_DISABLED")` instead -- a property of the build, which imports cannot change -- so the subprocess checks always run on a free-threaded interpreter. They were never the problem: a fresh child is unaffected by whatever the parent imported, which is why they still pass and still name the module responsible even in the broken state. The lost GIL is now reported by its own test rather than suppressing everything else. It is deliberately independent of the other checks, so a re-enabled GIL produces one specific failure instead of four vacuous skips. Verified in both directions on free-threaded 3.13.7: normal 5 passed (was 4 passed) PYTHON_GIL=1 1 failed, 4 passed, exit 1 (was 4 skipped, exit 0) Still a no-op on a GIL build (5 skipped), and the free-threaded job's full suite is unaffected: 2285 passed under PYTHON_GIL=0 with FAUST_REQUIRE_CYTHON=1. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HHM5xcCjiuyh3rPWFzEoP9 * Make the acknowledgement transition atomic `Message.ack` reads `acked`, decrements `refcount`, and on reaching zero runs the final-ack bookkeeping that marks an offset safe to commit. Those are separate steps with nothing holding them together, so two threads acking the same message can read the same refcount and both write `n - 1`. A decrement is lost, and the final ack then fires twice or never. `Event.ack()` is public API and nothing stops a caller invoking it from a thread, so this is reachable rather than theoretical. ## Not a free-threading bug, except where it is The developer guide recorded this as a race free-threading *would* expose, and left the fix open on the grounds that it would cost single-threaded users. That reading was half wrong, in both directions. The pure-Python path never needed free-threading: the GIL is released between bytecodes, and `self.refcount = self.refcount - n` is LOAD_ATTR / BINARY_OP / STORE_ATTR. With the switch interval turned down on GIL-enabled 3.11, 13 of 200 trials lost a decrement, and in 8 of 200 the final ack never ran at all -- an offset that never becomes committable. This has been reachable on every released faust. The compiled path is the opposite, and inverts the intuition that the extension is the riskier one. Compiled code does not return through the eval loop, so with a GIL held nothing switches threads inside `after()` and its transition is atomic by accident. 0 of 200 trials failed on 3.11. Take the GIL away and the accident goes: 6 of 50 trials lost an ack on 3.13t. ## The fix `ack_lock` serializes the whole transition rather than the decrement alone, across all three paths: `Message.ack`, `Consumer.ack`, and the Cython `after()`, which inlines the other two and so does not inherit their locking. Process-wide, not per-message, because the guarded state is: the final ack mutates `_acked_index`, `_acked`, `_n_acked` and `_unacked_messages`, shared by every message. A per-message lock would leave all of it exposed. Reentrant, because the transition nests -- `Message.ack` -> `ConsumerMessage.on_final_ack` -> `Consumer.ack`. On the cost that kept this open: faust acks from the event loop thread, so the ordinary case is one uncontended acquire against the dict and set operations the same section already performs. It is contended only when a caller acks from another thread, which is the case that was broken. ## Verification tests/unit/test_ack_concurrency.py covers all three paths, and each was confirmed by removing the lock and watching the specific test fail: Message.ack 13/200 trials lost an ack -> 0 final ack ran 0 times in 8/200 -> exactly once after() (3.13t) 6/50 trials lost an ack -> 0 Suites green in every configuration: 3.11 with extensions (2284 passed), without (2216 passed), free-threaded 3.13t under PYTHON_GIL=0 with FAUST_REQUIRE_CYTHON=1 (2289 passed). mypy clean over 165 files; flake8, isort and black clean. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HHM5xcCjiuyh3rPWFzEoP9 --------- Co-authored-by: Claude <noreply@anthropic.com> Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
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
Faust already ships optional Cython implementations of the window types (
faust/_cython/windows.pyx), the stream iterator (faust/_cython/streams.pyx) and the topic conductor (faust/transport/_cython/conductor.pyx). This adds three more, found by profiling the per-message and per-commit work.Each follows the pattern already in the tree: the pure-Python implementation stays and is used whenever the extension could not be built or
NO_CYTHONis set. Nothing here is required for Faust to run.What was added
faust/utils/_cython/functional.pyx—first_consecutive_run()Consumer._new_offsetonly needs the first run fromconsecutive_numbers(), but thegroupby()implementation builds a tuple, calls a Python key function and creates a group generator for every acked offset. It runs once per assigned partition on every commit, and because it blocks the event loop the cost shows up as latency rather than throughput.faust/transport/_cython/scheduler.pyx—records_iterator()A C-level cursor replacing the round-robin generator returned by
DefaultSchedulingStrategy.records_iterator, which also takes overTopicBuffer's inner per-partition generator. Two generator frames were being resumed for every record fetched from the broker.faust/sensors/_cython/base.pyx—SensorDelegateBaseA
cdefbase class carrying the four sensor hooks that fire on every message.SensorDelegatesubclasses it and keeps its ~20 occasional hooks in plain Python, so only the hot quarter of the class moved.Benchmarks
extra/tools/benchmark_cython.py(new) times each accelerator against its pure-Python counterpart in the same interpreter. On CPython 3.11:first_consecutive_run/ 100 offsetsfirst_consecutive_run/ 10k offsetsfirst_consecutive_run/ 100k offsetsrecords_iterator/ 1 topic × 1 partrecords_iterator/ 4 topics × 8 partsrecords_iterator/ 8 topics × 16 partsSensorDelegate.on_stream_event_in/ 1 sensorSensorDelegateall 4 hooks / 1 sensorThree honest caveats:
records_iteratorfigures were previously 5.7–7.7x here. That version skipped rebuilding its snapshot when the mapping length was unchanged, which turned out to drop records when a buffer was replaced under an existing key (see below). Correctness cost most of the win; these numbers are the fixed implementation. An unconditional rebuild was also measured, at 1.6–2.1x, so the current identity check is worth keeping over simply rebuilding.first_consecutive_run, part of the win is just droppinggroupby— the new pure-Python helper is already ~2.6x faster than the old expression, and Cython adds ~4-5x on top of that. The table compares the two new implementations, not old-vs-new.Monitorattached, the sensor bodies dominate and the relative win is smaller.Behaviour
Preserved rather than approximated. Specifically:
TopicBuffer._buffersis drained identically.TopicBuffersubclass is still driven throughnext(), so an overridden__iter__/__next__is honoured.map_from_recordsandrecords_iteratorremain overridable, so a customConsumerScheduleris unaffected.first_consecutive_runstops consuming as soon as the run ends, so a shared iterator is left exactly where the pure-Python version leaves it. It re-reads the list length while scanning, because a custom__sub__can resize the list underneath it._sensorsdirectly behaves the same in both.Not covered deliberately: swapping in a topic under a different name mid-iteration. The pure-Python generator iterates the live dict, so CPython raises
dictionary keys changed during iteration— undefined behaviour in Python itself rather than a guarantee to pin.Testing
NO_CYTHON=1(2305 passed), and extensions absent entirely.flake8/black/isortclean; sdist ships all six.pyxfiles and rebuilds from a clean tree.Evaluated and rejected
Recorded here so the ground does not get re-covered:
Message/ConsumerMessageascdefclasses —cdef publiccompiles to getset descriptors, which CPython 3.11+ cannot specialize the way it specializes the__slots__member descriptorsMessagealready uses. Most readers are pure Python, so attribute reads would likely regress. Only worth it bundled with atuples.pxdso the existing.pyxmodules cancimportit.Event—EventTisGeneric[T], AsyncContextManager, socdef class Event(EventT)does not compile. TheEventT.register(Event)workaround routes the per-messageisinstancechecks throughABCMeta.__instancecheck__, costing more than the constructor saves.Registry(serializers) — real dispatch waste, but most of it is recoverable in pure Python; Cython's marginal contribution is small.Schema,Codec,Table/Collection,Monitor,Consumer— all blocked byService/ABC bases, documented subclassing, or both.Recordcodegen andfaust/models/typing.py— generated viaexec()at class-definition time; Cython cannot apply.faust/utils/iso8601.py— already delegates tociso8601when installed.FieldDescriptor(faust/models/fields.py) is the one genuinely promising candidate left — ~30 descriptor calls per typed-model message — but it needs a hybridcdefbase plus converting twocached_propertyattributes, which is a larger change than this PR. Left as follow-up.Note on the assignor commit
2f04895is a cherry-pick of the existing fix onclaude/fix-pypy-hypothesis-deadline(which has no open PR). Without it thepypy3.11leg fails on a hypothesis deadline flake intest_copartitioned_assignor.pythat is unrelated to this diff — see the comment below for the evidence. Happy to drop it if that branch lands separately.🤖 Generated with Claude Code
https://claude.ai/code/session_019oX4oGCQGgvPJHabjwfaBk