Recover scheduler throughput lost to the snapshot-staleness check - #754
Open
wbarnha wants to merge 6 commits into
Open
Recover scheduler throughput lost to the snapshot-staleness check#754wbarnha wants to merge 6 commits into
wbarnha wants to merge 6 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
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
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
Fixing the record-loss bug in #751 cost most of the scheduler's speedup: the length-only shortcut it replaced was unsound, but the identity check that replaced it walks the mapping with .items(), allocating a view, an iterator and a tuple per entry. A single-partition topic reaches a pass boundary on every record, so that walk lands squarely on the hot path and the accelerator fell from 5.7-7.7x over pure Python to 2.2-2.8x. Two changes get it back without weakening the guarantee. TopicBuffer._buffers becomes a plain dict. It was an OrderedDict, with a comment noting it was "a regular dict, but ordered on Python 3.6" -- dicts have been insertion-ordered by language guarantee since 3.7 and Faust requires 3.10, so the subclass bought nothing. It also cost something: OrderedDict is not an exact dict, so it cannot be walked with PyDict_Next. Both _snapshot_is_current() methods then use PyDict_Next when the mapping is an exact dict, walking it with no allocation at all and comparing borrowed references by identity. Anything that is not an exact dict still goes through .items(), so a custom mapping handed to records_iterator keeps working. topology python cython before after 1t x 1p x 500 525.8n 90.2n 2.23x 5.83x 1t x 8p x 200 422.5n 78.2n -- 5.40x 4t x 8p x 100 336.6n 64.3n 2.63x 5.23x 8t x 16p x 50 324.8n 69.3n 2.77x 4.68x The correctness properties from #751 are unchanged and re-verified: both mid-iteration replacement scenarios still match the pure-Python iterator, and the randomised differential test is at 4000/4000. Adds tests for the non-dict mapping fallback and for _buffers being an exact dict, since that is now load-bearing rather than incidental. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019oX4oGCQGgvPJHabjwfaBk
wbarnha
changed the base branch from
claude/faust-cython-rewrites-vwltqs
to
master
August 12, 2026 10:33
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
Fixing the record-loss bug reported in review of #751 cost most of the scheduler's speedup. The length-only shortcut it replaced was unsound — it could not tell "unchanged" from "swapped" — but the identity check that replaced it walks the mapping with
.items(), allocating a view, an iterator and a tuple per entry. A single-partition topic reaches a pass boundary on every record, so that walk lands squarely on the hot path, and the accelerator fell from 5.7–7.7x over pure Python to 2.2–2.8x.Two changes get it back without weakening the guarantee.
TopicBuffer._buffersbecomes a plain dictIt was an
OrderedDict, carrying this comment:Dicts have been insertion-ordered by language guarantee since 3.7, and Faust requires 3.10, so the subclass bought nothing. It also cost something:
OrderedDictis not an exact dict, so it cannot be walked withPyDict_Next.Both
_snapshot_is_current()methods usePyDict_NextWhen the mapping is an exact dict, the check walks it with no allocation at all, comparing borrowed references by identity. Anything that is not an exact dict still goes through
.items(), so a custom mapping handed torecords_iteratorkeeps working.Benchmarks
ns per record, CPython 3.11:
Back to roughly where the unsound version was, with the correctness guarantee kept.
Correctness
Unchanged from #751 and re-verified rather than assumed:
TopicBufferswapped in under an existing topic name, and an iterator swapped in under an existingTP) still match the pure-Python iterator exactly.NO_CYTHON=1.New tests cover the two things this PR makes load-bearing:
.items()fallback;_buffersactually being an exactdict, sincePyDict_Nextsilently would not apply otherwise and the speedup would quietly disappear.Note on the
OrderedDictchangeThis is a small behavioural surface worth a reviewer's eye: anything relying on
_buffersbeing anOrderedDictspecifically —move_to_end,popitem(last=False), equality being order-sensitive — would be affected. Nothing in the tree does;_buffersis only assigned, popped and iterated. Flagging it because it is the one change here that is not purely internal to the.pyx.🤖 Generated with Claude Code
https://claude.ai/code/session_019oX4oGCQGgvPJHabjwfaBk
Generated by Claude Code