From 3f8605113c6a6a47e5732c31c646456f0ac4066a Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 4 Aug 2026 11:23:46 +0000 Subject: [PATCH 1/7] Evaluate a feature-flagged Rust accelerator build Assess whether faust should grow an optional, USE_RUST-gated Rust extension alongside the existing Cython ones, and record the result. A working USE_RUST build was prototyped against a clone of the repo (setuptools-rust + a PyO3 port of the Cython HoppingWindow) and benchmarked three ways. Findings: - Rust beats Cython only where a call does real work per crossing: ranges() 1.7x faster, but current()/stale()/earliest() are 13-17% *slower*, because the PyO3 call boundary costs 79ns against Cython's 59ns. - The two remaining Cython modules (streams, conductor) are async dispatch code whose bodies are calls back into Python -- the workload where that boundary cost is worst. - Every batch-shaped part of faust is already served by someone else's Rust: orjson, ciso8601, rocksdict (PyO3). So the recommendation is not to add a Rust build axis now. The document also records the integration details a future PR would need (MANIFEST.in misses *.rs, [build-system] requires cannot be conditional, RustExtension(optional=True) for graceful degradation, a worker-banner row), the CI cost (the pytest matrix doubles to 20 legs), and the concrete triggers that would change the answer. Ship the benchmark used, so the numbers can be re-checked on other hardware; it skips any implementation it cannot import. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01R9i6CXVTRGNRwEvzzCSP1B --- docs/proposals/rust-acceleration.md | 258 ++++++++++++++++++++++++++++ extra/tools/bench_accel_windows.py | 109 ++++++++++++ 2 files changed, 367 insertions(+) create mode 100644 docs/proposals/rust-acceleration.md create mode 100644 extra/tools/bench_accel_windows.py diff --git a/docs/proposals/rust-acceleration.md b/docs/proposals/rust-acceleration.md new file mode 100644 index 000000000..01cac5116 --- /dev/null +++ b/docs/proposals/rust-acceleration.md @@ -0,0 +1,258 @@ +# Evaluation: a feature-flagged Rust accelerator build for Faust + +Status: **evaluation only — no build changes are proposed for merge by this +document.** +Scope: whether Faust should grow an optional, feature-flagged Rust extension +alongside (or eventually instead of) its Cython extensions. + +**Recommendation in one line:** do not add a Rust build axis to Faust today. +The prototype works and the build integration is sound, but on the code Faust +actually accelerates, Rust is between 1.9x faster and 15% *slower* than the +Cython we already ship, and it would roughly double the test matrix and add a +second native toolchain for that. Prefer keeping Cython, and revisit only if a +genuinely batch-shaped hot path appears (see +[What would change this answer](#what-would-change-this-answer)). + +Everything below was measured or built, not estimated; see +[Reproducing](#reproducing). + +--- + +## 1. What "a feature-flagged build" would have to mean here + +Faust already has a working accelerator feature-flag pattern, and any Rust work +has to fit it rather than replace it: + +| Layer | Mechanism today | Location | +| --- | --- | --- | +| Build flag | `USE_CYTHON` / `NO_CYTHON` env vars | `setup.py:12-21` | +| Graceful build failure | `ve_build_ext` raises `BuildFailed`, `do_setup()` retried with no `ext_modules` | `setup.py:103-122`, `setup.py:216-222` | +| Runtime flag | `NO_CYTHON` env var re-read at import, plus `try: import ... except ImportError` fallback | `faust/streams.py:57-65`, `faust/windows.py:78-92`, `faust/transport/conductor.py:37-45` | +| Packaging extra | `faust[cython]`, rolled into `faust[fast]` | `requirements/extras/cython.txt`, `fast.txt` | +| Visibility | worker banner prints `+ Cython (compiler)` | `faust/cli/worker.py:156-163` | +| CI | `use-cython: ['true', 'false']` axis across 5 Pythons | `.github/workflows/python-package.yml` | + +A Rust build must reproduce **all six** rows, not just the first. A flag that +only gates compilation, without the import guard, the extra, the banner line +and the CI axis, produces a build that silently differs from what is tested. + +Worth noting up front: Faust *already ships Rust*, just not its own. Two +optional dependencies are Rust extension modules — `orjson` (declares +`Programming Language :: Rust`) and `rocksdict` (links `pyo3-0.27.1`). Both +arrive as prebuilt wheels and cost the project nothing. That is the cheap way +to consume Rust, and it is already being used. + +## 2. The prototype that was built and verified + +A working `USE_RUST` build was assembled against a clone of this repo (not on +this branch). It is small — this is the whole of it: + +**`setup.py`** (after the existing `cythonize` block): + +```python +def _flag(name, default=""): + v = os.environ.get(name, default) + return bool(v) and str(v).lower() not in {"0", "false", "no", "off"} + +USE_RUST = _flag("USE_RUST") +rust_extensions = [] +if USE_RUST: + try: + from setuptools_rust import Binding, RustExtension + except ImportError: + print("---*--- USE_RUST set but setuptools-rust missing: SKIPPING ---*---") + else: + print("---*--- USING RUST ---*---") + rust_extensions = [ + RustExtension( + "faust._rust._accel", + path="faust/_rust/Cargo.toml", + binding=Binding.PyO3, + py_limited_api=True, + optional=True, # a cargo failure must not fail the install + ) + ] +``` + +…passed through as `rust_extensions=rust_extensions` in `do_setup()`, plus +`"setuptools-rust>=1.9"` in `[build-system] requires`, plus +`faust/_rust/{Cargo.toml,src/lib.rs}` holding a PyO3 0.29 port of +`faust._cython.windows.HoppingWindow` (`abi3-py310`, `crate-type = ["cdylib"]`). + +Verified behaviours: + +* `USE_CYTHON=1 USE_RUST=1 pip install .` builds **both** accelerators in one + pass; `faust._rust._accel` imports and returns results identical to the + Python and Cython implementations for `ranges`, `current`, `stale`, + `earliest`. +* `py_limited_api=True` really does produce a stable-ABI object + (`_accel.abi3.so`), so *one* Rust build covers CPython 3.10–3.14. +* With `cargo` removed from `PATH` and `USE_RUST=1` still set, `pip install .` + **succeeds** and simply omits the module — `optional=True` gives Rust the + same graceful degradation `ve_build_ext` gives Cython. Without + `optional=True` this install aborts, which would be a regression for + source installs. +* Default (`USE_RUST` unset) builds are byte-for-byte unaffected and need no + Rust toolchain. + +Three integration details that are easy to miss and that any real PR must +handle: + +1. **`MANIFEST.in` excludes the Rust sources.** `recursive-include faust *.py + *.typed *.pyx` does not match `*.rs`, `Cargo.toml` or `Cargo.lock`, so the + sdist would ship without them and `USE_RUST=1` would be a no-op for anyone + installing from source. Needs an explicit include. +2. **`[build-system] requires` cannot be conditional.** PEP 518 has no env + switch, so `setuptools-rust` becomes an unconditional build dependency for + *everyone*, including the 99% who never set `USE_RUST`. It is pure Python + and small, but it is a new mandatory build-time download. +3. **`optional=True` fails silently.** A user who asks for `USE_RUST=1` and + gets no acceleration receives no signal at all. This is why the worker + banner row in the table above is not optional — a `_human_rust_info()` + sibling of `_human_cython_info()` (`faust/cli/worker.py:156`) is the only + way to tell the two builds apart at runtime. + +## 3. Measurements + +Linux x86_64, CPython 3.11, rustc 1.94.1, PyO3 0.29 (`--release`), Cython +extensions built with the project's standard `-O2`. `min` of 5 runs × +200 000 iterations, ns/call. Reproduce with +`extra/tools/bench_accel_windows.py`. + +### 3.1 `HoppingWindow`, the one accelerator with a like-for-like Rust port + +| case | python | cython | rust | cython vs py | rust vs py | **rust vs cython** | +| --- | ---: | ---: | ---: | ---: | ---: | ---: | +| `ranges(ts)` | 2045 ns | 1012 ns | 592 ns | 2.02x | 3.46x | **1.71x faster** | +| `current(ts)` | 425 ns | 151 ns | 174 ns | 2.81x | 2.44x | **0.87x — slower** | +| `stale(ts, ts+1)` | 559 ns | 136 ns | 157 ns | 4.11x | 3.57x | **0.87x — slower** | +| `earliest(ts)` | 357 ns | 141 ns | 170 ns | 2.53x | 2.10x | **0.83x — slower** | + +(An earlier run of the same benchmark put `ranges` at 1.91x and the three +scalar methods at 0.87–0.91x; run-to-run spread is a few percent, the sign is +stable.) + +The pattern is the important part, and it is not about Rust being slow: + +* **The win is proportional to work done per call.** `ranges` builds a list of + ~7 tuples and Rust wins 1.7x. The three methods that do a couple of + floating-point operations and return one tuple all *lose* to Cython. +* **Because the floor is the call boundary, not the language.** A bare + attribute read costs **60 ns through Cython and 79 ns through PyO3** — + PyO3's argument parsing, `Bound` handling and error plumbing are simply + thicker than `cdef class` access. Any call whose body is cheaper than ~30 ns + of that difference is a guaranteed loss, whatever the language. + +### 3.2 Build cost + +| build | wall time | +| --- | ---: | +| `pip install .` with `USE_CYTHON=1` | 12.5 s | +| `pip install .` with `USE_CYTHON=1 USE_RUST=1`, cold `CARGO_HOME` | 29.2 s | +| the Rust crate alone (`maturin build --release`, cold) | 21 s | + +So ~+17 s per cold build for a crate containing one 80-line struct — that is +almost entirely compiling the PyO3 macro stack (`syn`, `quote`, +`proc-macro2`, `pyo3-macros`), and it is a fixed cost that does not grow much +as Faust's own Rust grows. It is paid per CI leg, not once. + +## 4. Which Faust code could actually go to Rust + +| Candidate | Shape | Verdict | +| --- | --- | --- | +| `faust/_cython/windows.pyx` (109 lines) | Pure float math, no Python objects held | **Portable, mostly not worth it.** The port exists and is correct; §3.1 shows it wins only on `ranges`. | +| `faust/_cython/streams.pyx` (198 lines) | `async def next()`, awaits `chan_slow_get`, `maybe_async`, sensor callbacks | **Poor fit.** The body is an await-driven sequence of calls back into Python; PyO3 needs `pyo3-async-runtimes` to express it and pays the 79 ns boundary on *every* callback it drives. The Cython version wins here precisely because `cdef class` attribute access is cheap, which is the one thing Rust is worse at. | +| `faust/transport/_cython/conductor.pyx` (134 lines) | Same — async dispatch, all work is calling Python | **Poor fit**, same reason. | +| JSON encode/decode (`faust/utils/json.py`) | Genuinely batch-shaped | **Already solved** by `faust[orjson]`, which is Rust. Writing our own would be strictly worse. | +| ISO-8601 parsing (`faust/utils/_iso8601_python.py`) | Batch-shaped, small | **Already solved** by `faust[ciso8601]`. | +| State store | Batch-shaped, big | **Already solved** by `faust[rocksdict]`, which is PyO3. | +| Codec chains (`faust/serializers/codecs.py`) | Thin wrappers over `json`/`pickle`/`base64` | No meaningful compute of our own to move. | +| Model field coercion (`faust/models/`) | Plausible on paper | Deeply coupled to `typing` introspection and user-supplied Python callables; would cross the boundary constantly. Not evaluated further. | + +The summary of that table is the crux of the evaluation: **every part of Faust +whose shape suits Rust is already served by someone else's Rust**, and the +three things Faust accelerates itself are all latency-bound call-boundary code, +which is the one workload where Cython beats PyO3. + +## 5. What it would cost + +**CI matrix.** `test-pytest` runs 5 Pythons × `use-cython: [true, false]` = 10 +legs, plus 5 confluent legs and PyPy. A `use-rust` axis makes it 20, each also +paying the cargo build from §3.2 — and a Rust build that is never exercised +with `USE_RUST=1` in CI is worse than no Rust at all, because it would ship +untested. + +**Wheels.** `build_wheels` runs 4 runner images × cp310–cp314. Each needs Rust +inside the manylinux container (`before-all = "curl -sSf https://sh.rustup.rs | +sh -s -- -y"` plus a `PATH` entry in `environment`), and cargo re-runs per +Python version even though the `abi3` output is identical — roughly +6 minutes +across the release job. Tolerable; it is the test matrix, not the wheel job, +that hurts. + +**The abi3 upside does not materialise.** `abi3-py310` means one Rust build +serves every supported CPython — but only a package whose *every* extension is +`abi3` can ship one wheel per platform. Faust's Cython extensions are +version-specific, so wheels stay per-version and the stable-ABI win is +theoretical until Rust *replaces* Cython. That is a much larger project than a +feature flag, and §3.1 says it would make three of four window methods slower. + +**Platforms.** `[tool.cibuildwheel] skip` already drops musllinux and +free-threaded builds. Rust is neutral-to-positive here (PyO3 supports +free-threading; the Cython extension is the reason `cp31?t-*` is skipped), but +that is a benefit only for a replacement, not an addition. PyPy already runs +pure-Python (`USE_CYTHON: 'false'`), so a Rust module would simply be absent — +no new work. + +**A third copy of every accelerated semantic.** The Python and Cython window +implementations have *already* drifted: `_PyHoppingWindow(60, 10).expires` is +`None` while `faust._cython.windows.HoppingWindow(60, 10).expires` is `0.0`. +No test asserts the two implementations agree — the CI matrix flips +`USE_CYTHON` globally and runs the same tests, so a divergence only surfaces if +a test happens to touch it. Adding a third implementation triples that exposure +against a test suite that does not currently check for it. + +**Supply chain and maintenance.** ~14 transitive crates for a hello-world PyO3 +module, a `Cargo.lock` to keep current, an MSRV to track (Rust ≥1.64 pins +manylinux2014 as the floor — fine today), and a contributor base that must now +include someone who reads Rust. For a project this size that last one is the +real cost. + +## 6. What would change this answer + +Concrete triggers, in rough order of likelihood: + +1. **A batch-shaped hot path appears in Faust's own code** — something that + crosses the boundary once and then does thousands of operations + (windowed-table range scans over many keys, bulk changelog + reconstruction on recovery, a partition-wide sort/merge). §3.1 shows the + win scales with work-per-call; `ranges`, the most batch-like of four + methods, is the only one that wins. +2. **Profiling shows the Cython extensions are actually material.** Nobody has + published what fraction of Faust's per-message cost lives in + `streams.pyx` + `conductor.pyx`. If it is small, the entire accelerator + question — Rust or Cython — is the wrong thing to optimise. This is the + cheapest next step and should precede any language decision. +3. **Free-threaded CPython becomes a target.** If dropping the `cp31?t-*` skip + matters, a `Py_GIL_DISABLED`-safe accelerator is needed and PyO3 is a + better starting point than auditing the `.pyx` files. That argues for + *replacement*, and would need the §3.1 regressions solved first. +4. **Cython becomes a liability** — a Python release it does not support in + time, or the `cp3*`/free-threading friction already visible in + `pyproject.toml`'s skip comment. + +If (1) or (3) lands, the §2 prototype is the implementation: it is about 40 +lines of build wiring, it degrades correctly, and it has been shown to work. + +## 7. Reproducing + +```bash +python -m venv .venv && . .venv/bin/activate +pip install -e . # builds the Cython extensions +python extra/tools/bench_accel_windows.py # python vs cython (vs rust, if built) +``` + +The benchmark skips any implementation it cannot import, so it is useful on a +pure-Python install too. To reproduce the `rust` column, apply the §2 wiring, +port `faust/_cython/windows.pyx`'s `HoppingWindow` to +`faust/_rust/src/lib.rs`, and build with `USE_RUST=1 pip install -e .` — the +benchmark picks up `faust._rust._accel` automatically. diff --git a/extra/tools/bench_accel_windows.py b/extra/tools/bench_accel_windows.py new file mode 100644 index 000000000..92d88eb4c --- /dev/null +++ b/extra/tools/bench_accel_windows.py @@ -0,0 +1,109 @@ +#!/usr/bin/env python +"""Compare the available HoppingWindow implementations. + +Faust ships a pure-Python ``HoppingWindow`` and a Cython one, and the Rust +evaluation in ``docs/proposals/rust-acceleration.md`` adds a third candidate. +This script times whichever of them are importable, checks they agree, and +prints ns/call plus speedups so the numbers in that document can be re-checked +on other hardware. + +Usage:: + + pip install -e . # builds faust._cython.windows + python extra/tools/bench_accel_windows.py + +Any implementation that is not importable is simply skipped, so the script is +useful on a pure-Python install too. +""" + +import timeit +from typing import Any, Callable, Dict, List, Tuple + +from faust.windows import _PyHoppingWindow + +SIZE = 60.0 +STEP = 10.0 +EXPIRES = 3600.0 +TIMESTAMP = 1_700_000_000.123 +ITERATIONS = 200_000 +REPEAT = 5 + + +def _implementations() -> List[Tuple[str, Any]]: + impls: List[Tuple[str, Any]] = [("python", _PyHoppingWindow)] + try: + from faust._cython.windows import HoppingWindow as CythonHoppingWindow + except ImportError: + pass + else: + impls.append(("cython", CythonHoppingWindow)) + try: + # Not built by default; see docs/proposals/rust-acceleration.md. + from faust._rust._accel import HoppingWindow as RustHoppingWindow + except ImportError: + pass + else: + impls.append(("rust", RustHoppingWindow)) + return impls + + +CASES: Dict[str, Callable[[Any], Any]] = { + "ranges(ts)": lambda w: w.ranges(TIMESTAMP), + "current(ts)": lambda w: w.current(TIMESTAMP), + "stale(ts, ts+1)": lambda w: w.stale(TIMESTAMP, TIMESTAMP + 1.0), + "earliest(ts)": lambda w: w.earliest(TIMESTAMP), +} + + +def _time_ns_per_call(fn: Callable[[], Any]) -> float: + best = min(timeit.repeat(fn, number=ITERATIONS, repeat=REPEAT)) + return best / ITERATIONS * 1e9 + + +def main() -> None: + impls = _implementations() + names = [name for name, _ in impls] + print(f"implementations: {', '.join(names)}") + print(f"size={SIZE} step={STEP} expires={EXPIRES} " f"iterations={ITERATIONS}\n") + + windows = [(name, cls(SIZE, STEP, EXPIRES)) for name, cls in impls] + + for case, fn in CASES.items(): + results = {name: fn(window) for name, window in windows} + distinct = {repr(value) for value in results.values()} + if len(distinct) > 1: + print(f"MISMATCH in {case}: {results}") + + header = f"{'case':<18}" + "".join(f"{name:>12}" for name in names) + if "python" in names: + header += "".join( + f"{name + ' vs py':>14}" for name in names if name != "python" + ) + print(header) + + for case, fn in CASES.items(): + timings = { + name: _time_ns_per_call(lambda fn=fn, w=window: fn(w)) + for name, window in windows + } + row = f"{case:<18}" + "".join(f"{timings[n]:>10.0f}ns" for n in names) + if "python" in names: + base = timings["python"] + row += "".join( + f"{base / timings[n]:>13.2f}x" for n in names if n != "python" + ) + print(row) + + # Cost of crossing the Python/native boundary with no work behind it: the + # floor under any accelerator, and the reason fine-grained calls do not + # benefit from a faster language. + print() + for name, window in windows: + if name == "python": + continue + attr = _time_ns_per_call(lambda w=window: w.size) + print(f"{name:<8} attribute read (call-boundary floor): {attr:.0f}ns") + + +if __name__ == "__main__": + main() From 86391b84cadd25d531b95ab36a50267cf1f84c33 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 4 Aug 2026 12:19:35 +0000 Subject: [PATCH 2/7] Add three Cython accelerators: offsets, record scheduler, sensor fan-out 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 Claude-Session: https://claude.ai/code/session_019oX4oGCQGgvPJHabjwfaBk --- .gitignore | 2 + extra/tools/benchmark_cython.py | 256 ++++++++++++++++++++++++++ faust/sensors/_cython/__init__.py | 1 + faust/sensors/_cython/base.pyx | 67 +++++++ faust/sensors/base.py | 44 ++++- faust/transport/_cython/scheduler.pyx | 196 ++++++++++++++++++++ faust/transport/consumer.py | 4 +- faust/transport/utils.py | 56 ++++-- faust/utils/_cython/__init__.py | 1 + faust/utils/_cython/functional.pyx | 99 ++++++++++ faust/utils/functional.py | 41 ++++- setup.py | 21 +++ tests/unit/sensors/test_base.py | 92 +++++++++ tests/unit/transport/test_utils.py | 131 ++++++++++++- tests/unit/utils/test_functional.py | 107 ++++++++++- 15 files changed, 1084 insertions(+), 34 deletions(-) create mode 100644 extra/tools/benchmark_cython.py create mode 100644 faust/sensors/_cython/__init__.py create mode 100644 faust/sensors/_cython/base.pyx create mode 100644 faust/transport/_cython/scheduler.pyx create mode 100644 faust/utils/_cython/__init__.py create mode 100644 faust/utils/_cython/functional.pyx diff --git a/.gitignore b/.gitignore index e6974ba9a..be040cade 100644 --- a/.gitignore +++ b/.gitignore @@ -37,6 +37,8 @@ htmlcov/ *.so faust/_cython/*.c faust/transport/_cython/*.c +faust/utils/_cython/*.c +faust/sensors/_cython/*.c # virtualenvs /env diff --git a/extra/tools/benchmark_cython.py b/extra/tools/benchmark_cython.py new file mode 100644 index 000000000..67b760077 --- /dev/null +++ b/extra/tools/benchmark_cython.py @@ -0,0 +1,256 @@ +#!/usr/bin/env python3 +"""Micro-benchmark the optional Cython accelerators against pure Python. + +Faust ships a handful of hot code paths twice: a readable pure-Python +implementation, and a Cython one that is used instead whenever the extension +modules could be built (see ``NO_CYTHON``). This script imports *both* and +times them side by side in the same interpreter, so the numbers are directly +comparable. + +Usage: + +.. sourcecode:: console + + $ python setup.py build_ext --inplace + $ python extra/tools/benchmark_cython.py + +Anything that needs a running worker (the ``Stream`` iterator and the topic +``Conductor``) is out of scope here -- those are covered by the end-to-end +benchmark in ``extra/tools/benchmark.py``. +""" + +import sys +from time import perf_counter +from typing import Callable, List, Optional, Tuple + +from faust.sensors.base import _PySensorDelegateBase, _SensorDelegateBase +from faust.transport.utils import ( + DefaultSchedulingStrategy, + _py_records_iterator, + _records_iterator, +) +from faust.types import TP +from faust.utils.functional import _py_first_consecutive_run, first_consecutive_run +from faust.windows import ( + HoppingWindow, + SlidingWindow, + _PyHoppingWindow, + _PySlidingWindow, +) + +MIN_TIME = 0.2 # seconds to spend on each timed run +Case = Tuple[str, Callable[[], None], Callable[[], None], int] + + +def _time(fn: Callable[[], None], iterations: int) -> float: + start = perf_counter() + for _ in range(iterations): + fn() + return perf_counter() - start + + +def _autorange(fn: Callable[[], None]) -> int: + """Find an iteration count that runs for at least ``MIN_TIME``.""" + iterations = 1 + while True: + elapsed = _time(fn, iterations) + if elapsed >= MIN_TIME or iterations >= 1 << 30: + return iterations + iterations *= 4 if elapsed < MIN_TIME / 8 else 2 + + +def _best(fn: Callable[[], None], iterations: int, rounds: int = 5) -> float: + """Return the best per-operation time in nanoseconds.""" + return min(_time(fn, iterations) for _ in range(rounds)) / iterations * 1e9 + + +def run(cases: List[Case]) -> bool: + """Time each case and print a comparison table.""" + name_width = max(len(name) for name, _, _, _ in cases) + print( + f"{'benchmark'.ljust(name_width)} {'python':>12} " + f"{'cython':>12} {'speedup':>8}" + ) + print("-" * (name_width + 38)) + + all_faster = True + for name, py_fn, cy_fn, scale in cases: + py_fn() + cy_fn() + iterations = _autorange(cy_fn) + py_ns = _best(py_fn, max(1, iterations // 4)) / scale + cy_ns = _best(cy_fn, iterations) / scale + speedup = py_ns / cy_ns + all_faster &= speedup > 1.0 + print( + f"{name.ljust(name_width)} {py_ns:9.1f} ns " + f"{cy_ns:9.1f} ns {speedup:7.2f}x" + ) + return all_faster + + +def offset_cases() -> List[Case]: + """faust.utils.functional.first_consecutive_run -- once per TP per commit.""" + cases = [] + for size in (100, 10_000, 100_000): + acked = list(range(size)) + cases.append( + ( + f"first_consecutive_run/{size} offsets", + lambda acked=acked: _py_first_consecutive_run(acked), + lambda acked=acked: first_consecutive_run(acked), + 1, + ) + ) + return cases + + +def _drain(impl: Callable, records: dict) -> None: + for _ in impl(DefaultSchedulingStrategy.map_from_records(records)): + pass + + +def scheduler_cases() -> List[Case]: + """faust.transport.utils records_iterator -- once per fetched record.""" + cases = [] + for topics, partitions, per_partition in ((1, 1, 500), (4, 8, 100), (8, 16, 50)): + records = { + TP(f"topic-{t}", p): list(range(per_partition)) + for t in range(topics) + for p in range(partitions) + } + cases.append( + ( + f"records_iterator/{topics}t x {partitions}p x {per_partition}", + lambda records=records: _drain(_py_records_iterator, records), + lambda records=records: _drain(_records_iterator, records), + topics * partitions * per_partition, + ) + ) + return cases + + +class _NoopSensor: + """Stand-in sensor: measures delegation overhead, not sensor work.""" + + beacon = None + + def on_message_in(self, tp, offset, message) -> None: ... + + def on_stream_event_in(self, tp, offset, stream, event) -> None: + return None + + def on_stream_event_out(self, tp, offset, stream, event, state=None) -> None: ... + + def on_message_out(self, tp, offset, message) -> None: ... + + +class _FakeBeacon: + def new(self, sensor: object) -> None: + return None + + +class _FakeApp: + beacon = _FakeBeacon() + + +_TP = TP("t", 0) +_OFFSET = 42 + + +def _event_in(delegate: object) -> None: + delegate.on_stream_event_in(_TP, _OFFSET, None, None) + + +def _all_hooks(delegate: object) -> None: + """The exact sequence of delegate calls a single message triggers.""" + delegate.on_message_in(_TP, _OFFSET, None) + state = delegate.on_stream_event_in(_TP, _OFFSET, None, None) + delegate.on_stream_event_out(_TP, _OFFSET, None, None, state) + delegate.on_message_out(_TP, _OFFSET, None) + + +def sensor_cases() -> List[Case]: + """faust.sensors.base -- four hooks fire on every single message.""" + cases = [] + for n_sensors in (1, 3): + py = _PySensorDelegateBase(_FakeApp()) + cy = _SensorDelegateBase(_FakeApp()) + for _ in range(n_sensors): + py.add(_NoopSensor()) + cy.add(_NoopSensor()) + + cases.extend( + [ + ( + f"SensorDelegate.on_stream_event_in/{n_sensors} sensor(s)", + lambda py=py: _event_in(py), + lambda cy=cy: _event_in(cy), + 1, + ), + ( + f"SensorDelegate all 4 hooks/{n_sensors} sensor(s)", + lambda py=py: _all_hooks(py), + lambda cy=cy: _all_hooks(cy), + 1, + ), + ] + ) + return cases + + +def window_cases() -> List[Case]: + """faust.windows -- already shipped; included for completeness.""" + py_hopping = _PyHoppingWindow(10.0, 5.0, 60.0) + cy_hopping = HoppingWindow(10.0, 5.0, 60.0) + py_sliding = _PySlidingWindow(10.0, 5.0, 60.0) + cy_sliding = SlidingWindow(10.0, 5.0, 60.0) + timestamp = 1_600_000_000.0 + return [ + ( + "HoppingWindow.ranges", + lambda: py_hopping.ranges(timestamp), + lambda: cy_hopping.ranges(timestamp), + 1, + ), + ( + "HoppingWindow.current", + lambda: py_hopping.current(timestamp), + lambda: cy_hopping.current(timestamp), + 1, + ), + ( + "SlidingWindow.ranges", + lambda: py_sliding.ranges(timestamp), + lambda: cy_sliding.ranges(timestamp), + 1, + ), + ] + + +def main(argv: Optional[List[str]] = None) -> int: + """Run every benchmark group.""" + if _records_iterator is _py_records_iterator: + print( + "The Cython extensions are not built (or NO_CYTHON is set), so " + "there is nothing to compare against.\n" + "Build them first with: python setup.py build_ext --inplace", + file=sys.stderr, + ) + return 1 + + groups = [ + ("offset commit path", offset_cases), + ("consumer record scheduler", scheduler_cases), + ("sensor delegation", sensor_cases), + ("windows", window_cases), + ] + for title, make_cases in groups: + print(f"\n== {title} ==") + run(make_cases()) + print("\nLower is better; speedup is python/cython.") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/faust/sensors/_cython/__init__.py b/faust/sensors/_cython/__init__.py new file mode 100644 index 000000000..668033668 --- /dev/null +++ b/faust/sensors/_cython/__init__.py @@ -0,0 +1 @@ +"""Cython optimized sensor components.""" diff --git a/faust/sensors/_cython/base.pyx b/faust/sensors/_cython/base.pyx new file mode 100644 index 000000000..72a960631 --- /dev/null +++ b/faust/sensors/_cython/base.pyx @@ -0,0 +1,67 @@ +# cython: language_level=3 +"""Cython optimized sensor delegation.""" + + +cdef class SensorDelegateBase: + """Per-message half of :class:`faust.sensors.base.SensorDelegate`. + + Only the four hooks that run on every message live here; the rest stay + in the Python subclass, which keeps them readable and overridable. + + The sensor set is read live, exactly as the pure-Python implementation + does, so mutating ``_sensors`` directly behaves the same either way. + """ + + cdef public object app + cdef public set _sensors + + def __init__(self, object app): + self.app = app + self._sensors = set() + + def add(self, sensor): + """Add sensor.""" + # connect beacons + sensor.beacon = self.app.beacon.new(sensor) + self._sensors.add(sensor) + + def remove(self, sensor): + """Remove sensor.""" + self._sensors.remove(sensor) + + def __iter__(self): + return iter(self._sensors) + + def on_message_in(self, object tp, object offset, object message): + """Call before message is delegated to streams.""" + cdef object sensor + for sensor in self._sensors: + sensor.on_message_in(tp, offset, message) + + def on_stream_event_in(self, object tp, object offset, object stream, + object event): + """Call when stream starts processing an event.""" + cdef: + dict states = {} + object sensor + for sensor in self._sensors: + states[sensor] = sensor.on_stream_event_in(tp, offset, stream, event) + return states + + def on_stream_event_out(self, object tp, object offset, object stream, + object event, object state=None): + """Call when stream is done processing an event.""" + cdef object sensor + if state: + for sensor in self._sensors: + sensor.on_stream_event_out( + tp, offset, stream, event, state.get(sensor)) + else: + for sensor in self._sensors: + sensor.on_stream_event_out(tp, offset, stream, event, None) + + def on_message_out(self, object tp, object offset, object message): + """Call when message is fully acknowledged and can be committed.""" + cdef object sensor + for sensor in self._sensors: + sensor.on_message_out(tp, offset, message) diff --git a/faust/sensors/base.py b/faust/sensors/base.py index 46f679e67..90127349c 100644 --- a/faust/sensors/base.py +++ b/faust/sensors/base.py @@ -1,5 +1,6 @@ """Base-interface for sensors.""" +import os from time import monotonic from typing import Any, Dict, Iterator, Mapping, Optional, Set @@ -14,6 +15,8 @@ __all__ = ["Sensor", "SensorDelegate"] +NO_CYTHON = bool(os.environ.get("NO_CYTHON", False)) + class Sensor(SensorT, Service): """Base class for sensors. @@ -151,8 +154,13 @@ def asdict(self) -> Mapping: return {} -class SensorDelegate(SensorDelegateT): - """A class that delegates sensor methods to a list of sensors.""" +class _PySensorDelegateBase: + """Pure-Python implementation of the per-message sensor fan-out. + + Split out from :class:`SensorDelegate` so the four hooks that run on + every single message can be swapped for a Cython implementation while + the (much larger) set of occasional hooks stays plain Python. + """ _sensors: Set[SensorT] @@ -197,16 +205,38 @@ def on_stream_event_out( tp, offset, stream, event, sensor_state.get(sensor) ) - def on_topic_buffer_full(self, tp: TP) -> None: - """Call when conductor topic buffer is full and has to wait.""" - for sensor in self._sensors: - sensor.on_topic_buffer_full(tp) - def on_message_out(self, tp: TP, offset: int, message: Message) -> None: """Call when message is fully acknowledged and can be committed.""" for sensor in self._sensors: sensor.on_message_out(tp, offset, message) + +if not NO_CYTHON: # pragma: no cover + try: + from ._cython.base import SensorDelegateBase as _SensorDelegateBase + except ImportError: + _SensorDelegateBase = _PySensorDelegateBase # type: ignore[misc,assignment] +else: # pragma: no cover + _SensorDelegateBase = _PySensorDelegateBase # type: ignore[misc,assignment] + + +class SensorDelegate( # type: ignore[misc,valid-type] + _SensorDelegateBase, + SensorDelegateT, +): + """A class that delegates sensor methods to a list of sensors. + + The four per-message hooks (``on_message_in``, ``on_stream_event_in``, + ``on_stream_event_out`` and ``on_message_out``) come from + :data:`_SensorDelegateBase`, which is the Cython implementation when it + could be built and :class:`_PySensorDelegateBase` otherwise. + """ + + def on_topic_buffer_full(self, tp: TP) -> None: + """Call when conductor topic buffer is full and has to wait.""" + for sensor in self._sensors: + sensor.on_topic_buffer_full(tp) + def on_table_get(self, table: CollectionT, key: Any) -> None: """Call when value in table is retrieved.""" for sensor in self._sensors: diff --git a/faust/transport/_cython/scheduler.pyx b/faust/transport/_cython/scheduler.pyx new file mode 100644 index 000000000..5d153c02d --- /dev/null +++ b/faust/transport/_cython/scheduler.pyx @@ -0,0 +1,196 @@ +# cython: language_level=3 +"""Cython optimized consumer record scheduler.""" +from cpython.list cimport PyList_GET_ITEM, PyList_GET_SIZE + + +cdef object _SENTINEL = object() + +# Resolved lazily: faust.transport.utils imports this module at the bottom of +# its own body, so importing it back at module scope here would be circular. +cdef object _TOPIC_BUFFER = None + + +cdef inline object _topic_buffer_type(): + global _TOPIC_BUFFER + if _TOPIC_BUFFER is None: + from faust.transport.utils import TopicBuffer + _TOPIC_BUFFER = TopicBuffer + return _TOPIC_BUFFER + + +cdef class _TopicCursor: + """Round-robin cursor over the partition buffers of a single topic. + + A C-level rewrite of :meth:`faust.transport.utils.TopicBuffer.__iter__`: + one record is taken from each partition in turn, and partitions that run + dry are popped from the buffer map at the end of the current pass. + + Anything that is not a plain, untouched ``TopicBuffer`` is driven through + ``next()`` instead, so a subclass that overrides ``__iter__``/``__next__`` + keeps working. + """ + + cdef: + object source # the TopicBuffer, kept so we can detect replacement + object buffers # TopicBuffer._buffers, read live + object compat # set instead of `buffers` for non-TopicBuffer entries + set to_remove + list tps + list iters + Py_ssize_t pi + Py_ssize_t n + + def __cinit__(self, object buffer): + self.source = buffer + self.buffers = None + self.compat = None + self.to_remove = set() + self.tps = [] + self.iters = [] + self.pi = 0 + self.n = 0 + if type(buffer) is _topic_buffer_type() and (buffer)._it is None: + # Untouched TopicBuffer: drive its partition iterators directly, + # which takes its generator out of the per-record path. + self.buffers = (buffer)._buffers + else: + self.compat = buffer + + cdef object next(self): + """Return the next ``(tp, record)`` pair, or the sentinel if drained.""" + cdef: + Py_ssize_t i + object item + + if self.compat is not None: + return next(self.compat, _SENTINEL) + + while True: + if self.pi >= self.n: + # `while buffers:` in the pure-Python generator. + if not self.buffers: + return _SENTINEL + self._start_pass() + if self.n == 0: + return _SENTINEL + i = self.pi + self.pi += 1 + item = next(PyList_GET_ITEM(self.iters, i), _SENTINEL) + if item is _SENTINEL: + self.to_remove.add(PyList_GET_ITEM(self.tps, i)) + continue + return (PyList_GET_ITEM(self.tps, i), item) + + cdef _start_pass(self): + cdef: + object tp + object it + + if self.to_remove: + for tp in self.to_remove: + self.buffers.pop(tp, None) + self.to_remove.clear() + elif PyList_GET_SIZE(self.tps) == len(self.buffers): + # Nothing drained and nothing added since the last pass, so the + # snapshot is still accurate. TopicBuffer.add() asserts the + # partition is new, so the size can only change on a real change. + self.pi = 0 + return + self.tps = [] + self.iters = [] + for tp, it in self.buffers.items(): + self.tps.append(tp) + self.iters.append(it) + self.n = PyList_GET_SIZE(self.tps) + self.pi = 0 + + +cdef class RoundRobinRecordIterator: + """Iterate a topic index map in round-robin order. + + A C-level rewrite of the generator returned by + :meth:`faust.transport.utils.DefaultSchedulingStrategy.records_iterator`, + flattening both round-robins (across topics, and across the partitions + within a topic) so no generator frame is resumed per record. + + The index map and each topic's buffer map are re-read at the start of + every pass, and drained entries are popped from them, exactly as the + pure-Python version does. + """ + + cdef: + object index + set to_remove + dict cursors + list topics + list topic_cursors + Py_ssize_t ti + Py_ssize_t n + + def __cinit__(self, object index): + self.index = index + self.to_remove = set() + self.cursors = {} + self.topics = [] + self.topic_cursors = [] + self.ti = 0 + self.n = 0 + + def __iter__(self): + return self + + def __next__(self): + cdef: + Py_ssize_t i + _TopicCursor cursor + object item + + while True: + if self.ti >= self.n: + # `while index:` in the pure-Python generator. + if not self.index: + raise StopIteration() + self._start_pass() + if self.n == 0: + raise StopIteration() + i = self.ti + self.ti += 1 + cursor = <_TopicCursor>PyList_GET_ITEM(self.topic_cursors, i) + item = cursor.next() + if item is _SENTINEL: + # This topic is now empty, but we cannot remove it from the + # map while iterating over it, so it goes to the next pass. + self.to_remove.add(PyList_GET_ITEM(self.topics, i)) + continue + return item + + cdef _start_pass(self): + cdef: + object topic + object buffer + _TopicCursor cursor + + if self.to_remove: + for topic in self.to_remove: + self.index.pop(topic, None) + self.cursors.pop(topic, None) + self.to_remove.clear() + elif PyList_GET_SIZE(self.topics) == len(self.index): + self.ti = 0 + return + self.topics = [] + self.topic_cursors = [] + for topic, buffer in self.index.items(): + cursor = <_TopicCursor>self.cursors.get(topic) + if cursor is None or cursor.source is not buffer: + cursor = _TopicCursor(buffer) + self.cursors[topic] = cursor + self.topics.append(topic) + self.topic_cursors.append(cursor) + self.n = PyList_GET_SIZE(self.topics) + self.ti = 0 + + +cpdef object records_iterator(object index): + """Iterate over a topic index map in round-robin order.""" + return RoundRobinRecordIterator(index) diff --git a/faust/transport/consumer.py b/faust/transport/consumer.py index d5a82b1b8..aef99733e 100644 --- a/faust/transport/consumer.py +++ b/faust/transport/consumer.py @@ -95,7 +95,7 @@ ) from faust.types.tuples import FutureMessage from faust.utils import terminal -from faust.utils.functional import consecutive_numbers +from faust.utils.functional import first_consecutive_run from faust.utils.tracing import traced_from_parent_span if typing.TYPE_CHECKING: # pragma: no cover @@ -1117,7 +1117,7 @@ def _new_offset(self, tp: TP) -> Optional[int]: # Note: acked is always kept sorted. # find first list of consecutive numbers - batch = next(consecutive_numbers(acked)) + batch = first_consecutive_run(acked) # remove them from the list to clean up. acked[: len(batch)] = [] self._acked_index[tp].difference_update(batch) diff --git a/faust/transport/utils.py b/faust/transport/utils.py index dc0b8d7d0..fe3f7805a 100644 --- a/faust/transport/utils.py +++ b/faust/transport/utils.py @@ -1,5 +1,6 @@ """Transport utils - scheduling.""" +import os from collections import OrderedDict from typing import ( Any, @@ -22,6 +23,8 @@ "TopicBuffer", ] +NO_CYTHON = bool(os.environ.get("NO_CYTHON", False)) + # But we want to process records from topics in round-robin order. # We convert records into a mapping from topic-name to "chain-of-buffers": # topic_index['topic-name'] = chain(all_topic_partition_buffers) @@ -54,22 +57,27 @@ def iterate(self, records: Mapping[TP, List]) -> Iterator[Tuple[TP, Any]]: def records_iterator(self, index: TopicIndexMap) -> Iterator[Tuple[TP, Any]]: """Iterate over topic index map in round-robin order.""" - to_remove: Set[str] = set() - sentinel = object() - _next = next - while index: - for topic in to_remove: - index.pop(topic, None) - for topic, messages in index.items(): - item = _next(messages, sentinel) - if item is sentinel: - # this topic is now empty, - # but we cannot remove from dict while iterating over it, - # so move that to the outer loop. - to_remove.add(topic) - continue - tp, record = item # type: ignore - yield tp, record + return _records_iterator(index) + + +def _py_records_iterator(index: TopicIndexMap) -> Iterator[Tuple[TP, Any]]: + """Iterate over topic index map in round-robin order.""" + to_remove: Set[str] = set() + sentinel = object() + _next = next + while index: + for topic in to_remove: + index.pop(topic, None) + for topic, messages in index.items(): + item = _next(messages, sentinel) + if item is sentinel: + # this topic is now empty, + # but we cannot remove from dict while iterating over it, + # so move that to the outer loop. + to_remove.add(topic) + continue + tp, record = item # type: ignore + yield tp, record class TopicBuffer(Iterator): @@ -115,3 +123,19 @@ def __next__(self) -> Tuple[TP, Any]: if it is None: it = self._it = iter(self) return it.__next__() + + +# The Cython version flattens both round-robins (across topics, and across the +# partitions within a topic) into a single C-level cursor, so no generator +# frame has to be resumed for each record fetched from the broker. +# +# Only the default iteration strategy is swapped: ``map_from_records`` and +# ``records_iterator`` remain overridable, and a ``TopicBuffer`` subclass falls +# back to being driven through ``next()``. +if not NO_CYTHON: # pragma: no cover + try: + from ._cython.scheduler import records_iterator as _records_iterator + except ImportError: + _records_iterator = _py_records_iterator +else: # pragma: no cover + _records_iterator = _py_records_iterator diff --git a/faust/utils/_cython/__init__.py b/faust/utils/_cython/__init__.py new file mode 100644 index 000000000..52d726339 --- /dev/null +++ b/faust/utils/_cython/__init__.py @@ -0,0 +1 @@ +"""Cython optimized functional utilities.""" diff --git a/faust/utils/_cython/functional.pyx b/faust/utils/_cython/functional.pyx new file mode 100644 index 000000000..70c3e4edf --- /dev/null +++ b/faust/utils/_cython/functional.pyx @@ -0,0 +1,99 @@ +# cython: language_level=3 +"""Cython optimized functional utilities.""" +from cpython.list cimport PyList_GET_ITEM, PyList_GET_SIZE +from cpython.long cimport PyLong_AsLongLongAndOverflow, PyLong_CheckExact + + +cdef object _SENTINEL = object() + + +cpdef list first_consecutive_run(object numbers): + """Return the first run of consecutive numbers in ``numbers``. + + Equivalent to ``next(consecutive_numbers(numbers), [])``, but without + building the intermediate tuples, the per-element Python key function + and the group generators that :func:`itertools.groupby` needs. + + A run continues for as long as each number is exactly one greater than + the one before it, which is the same rule ``groupby`` applies when + grouping on ``index - value``. Repeated numbers therefore end a run. + + Like the pure-Python version this stops consuming as soon as the run + ends, so a non-sequence iterable is left positioned just past the first + number that broke the run. + """ + if type(numbers) is list: + return _run_from_list(numbers) + return _run_from_iterable(iter(numbers)) + + +cdef list _run_from_list(list seq): + cdef: + list run + Py_ssize_t i + object prev + object cur + long long c_prev + long long c_cur + int overflow + + if PyList_GET_SIZE(seq) == 0: + return [] + + prev = PyList_GET_ITEM(seq, 0) + run = [prev] + i = 1 + + # Fast path: plain ints that fit in a C long long. Kafka offsets always + # do, so this is what actually runs in production. Nothing in this loop + # can run Python code, so the size only has to be re-read for the slow + # path below. + if PyLong_CheckExact(prev): + c_prev = PyLong_AsLongLongAndOverflow(prev, &overflow) + if not overflow: + while i < PyList_GET_SIZE(seq): + cur = PyList_GET_ITEM(seq, i) + if not PyLong_CheckExact(cur): + break + c_cur = PyLong_AsLongLongAndOverflow(cur, &overflow) + # ``c_cur <= c_prev`` is tested first so the subtraction + # below can never underflow. + if overflow or c_cur <= c_prev or c_cur - 1 != c_prev: + break + run.append(cur) + c_prev = c_cur + i += 1 + if i >= PyList_GET_SIZE(seq): + return run + prev = PyList_GET_ITEM(seq, i - 1) + + # Slow path: arbitrary objects supporting ``-`` and comparison to 1. + # ``cur - prev`` runs arbitrary Python code that may resize the list, + # so the length is re-read on every iteration. + while i < PyList_GET_SIZE(seq): + cur = PyList_GET_ITEM(seq, i) + if cur - prev != 1: + break + run.append(cur) + prev = cur + i += 1 + return run + + +cdef list _run_from_iterable(object it): + cdef: + list run + object prev + object cur + + prev = next(it, _SENTINEL) + if prev is _SENTINEL: + return [] + run = [prev] + while True: + cur = next(it, _SENTINEL) + if cur is _SENTINEL or cur - prev != 1: + break + run.append(cur) + prev = cur + return run diff --git a/faust/utils/functional.py b/faust/utils/functional.py index 98d4ad764..be3eada26 100644 --- a/faust/utils/functional.py +++ b/faust/utils/functional.py @@ -1,16 +1,20 @@ """Functional utilities.""" +import os from functools import reduce from itertools import groupby -from typing import Iterable, Iterator, Mapping, Sequence, Tuple, TypeVar +from typing import Iterable, Iterator, List, Mapping, Sequence, Tuple, TypeVar __all__ = [ "consecutive_numbers", + "first_consecutive_run", "translate", ] T = TypeVar("T") +NO_CYTHON = bool(os.environ.get("NO_CYTHON", False)) + def consecutive_numbers(it: Iterable[int]) -> Iterator[Sequence[int]]: """Find runs of consecutive numbers. @@ -22,6 +26,41 @@ def consecutive_numbers(it: Iterable[int]) -> Iterator[Sequence[int]]: yield [a[1] for a in g] +def _py_first_consecutive_run(numbers: Iterable[int]) -> List[int]: + """Return the first run of consecutive numbers in ``numbers``. + + Equivalent to ``next(consecutive_numbers(numbers), [])``, but without + building the intermediate tuples, the per-element key function and the + group generators that :func:`itertools.groupby` needs. + + Callers that only need the first run (such as the consumer working out + the next offset to commit) should use this instead of + :func:`consecutive_numbers`, as it is the part of the commit path that + scales with the number of un-committed offsets. + """ + it = iter(numbers) + try: + prev = next(it) + except StopIteration: + return [] + run = [prev] + for cur in it: + if cur - prev != 1: + break + run.append(cur) + prev = cur + return run + + +if not NO_CYTHON: # pragma: no cover + try: + from ._cython.functional import first_consecutive_run + except ImportError: + first_consecutive_run = _py_first_consecutive_run +else: # pragma: no cover + first_consecutive_run = _py_first_consecutive_run + + def translate(table: Mapping, s: str) -> str: """Replace characters and patterns in string ``s``. diff --git a/setup.py b/setup.py index 7a3190b70..003c4cc56 100644 --- a/setup.py +++ b/setup.py @@ -88,6 +88,27 @@ extra_compile_args=CFLAGS, extra_link_args=LDFLAGS, ), + Extension( + "faust.sensors._cython.base", + ["faust/sensors/_cython/base" + ext], + libraries=LIBRARIES, + extra_compile_args=CFLAGS, + extra_link_args=LDFLAGS, + ), + Extension( + "faust.transport._cython.scheduler", + ["faust/transport/_cython/scheduler" + ext], + libraries=LIBRARIES, + extra_compile_args=CFLAGS, + extra_link_args=LDFLAGS, + ), + Extension( + "faust.utils._cython.functional", + ["faust/utils/_cython/functional" + ext], + libraries=LIBRARIES, + extra_compile_args=CFLAGS, + extra_link_args=LDFLAGS, + ), ] diff --git a/tests/unit/sensors/test_base.py b/tests/unit/sensors/test_base.py index ad96d9e6e..ce0ce74aa 100644 --- a/tests/unit/sensors/test_base.py +++ b/tests/unit/sensors/test_base.py @@ -5,6 +5,7 @@ from faust import Event, Stream, Table, Topic, web from faust.assignor import PartitionAssignor from faust.sensors import Sensor +from faust.sensors.base import _PySensorDelegateBase, _SensorDelegateBase from faust.transport.consumer import Consumer from faust.transport.producer import Producer from faust.types import TP, Message @@ -239,3 +240,94 @@ def test_on_web_request(self, *, sensors, sensor, app, req, response, view): def test_repr(self, *, sensors): assert repr(sensors) + + +#: Both implementations of the per-message sensor fan-out. +#: ``_SensorDelegateBase`` is the Cython one whenever the extension could be +#: built, and is otherwise the same object as ``_PySensorDelegateBase``. +SENSOR_DELEGATE_BASES = [_PySensorDelegateBase, _SensorDelegateBase] + + +class Test_SensorDelegateBase: + """The four hooks that run on every message, in both implementations.""" + + def _delegate(self, base, n_sensors=1): + app = Mock(name="app") + delegate = base(app) + sensors = [] + for _ in range(n_sensors): + sensor = Mock(name="sensor", autospec=Sensor) + delegate.add(sensor) + sensors.append(sensor) + return delegate, sensors + + @pytest.mark.parametrize("base", SENSOR_DELEGATE_BASES) + def test_add_connects_beacon(self, base): + delegate, [sensor] = self._delegate(base) + assert sensor.beacon is delegate.app.beacon.new.return_value + assert list(delegate) == [sensor] + + @pytest.mark.parametrize("base", SENSOR_DELEGATE_BASES) + def test_remove(self, base): + delegate, [sensor] = self._delegate(base) + delegate.remove(sensor) + assert not list(delegate) + delegate.on_message_in(TP1, 3, None) + sensor.on_message_in.assert_not_called() + + @pytest.mark.parametrize("base", SENSOR_DELEGATE_BASES) + def test_remove__missing_raises(self, base): + delegate, _ = self._delegate(base) + with pytest.raises(KeyError): + delegate.remove(Mock(name="never-added")) + + @pytest.mark.parametrize("base", SENSOR_DELEGATE_BASES) + def test_no_sensors(self, base): + delegate = base(Mock(name="app")) + delegate.on_message_in(TP1, 3, None) + assert delegate.on_stream_event_in(TP1, 3, None, None) == {} + delegate.on_stream_event_out(TP1, 3, None, None, None) + delegate.on_message_out(TP1, 3, None) + + @pytest.mark.parametrize("base", SENSOR_DELEGATE_BASES) + @pytest.mark.parametrize("n_sensors", [1, 3]) + def test_on_message_in_out(self, base, n_sensors, message): + delegate, sensors = self._delegate(base, n_sensors) + delegate.on_message_in(TP1, 303, message) + delegate.on_message_out(TP1, 303, message) + for sensor in sensors: + sensor.on_message_in.assert_called_once_with(TP1, 303, message) + sensor.on_message_out.assert_called_once_with(TP1, 303, message) + + @pytest.mark.parametrize("base", SENSOR_DELEGATE_BASES) + @pytest.mark.parametrize("n_sensors", [1, 3]) + def test_on_stream_event_in_out(self, base, n_sensors, stream, event): + delegate, sensors = self._delegate(base, n_sensors) + state = delegate.on_stream_event_in(TP1, 303, stream, event) + assert set(state) == set(sensors) + delegate.on_stream_event_out(TP1, 303, stream, event, state) + for sensor in sensors: + sensor.on_stream_event_in.assert_called_once_with(TP1, 303, stream, event) + sensor.on_stream_event_out.assert_called_once_with( + TP1, 303, stream, event, state[sensor] + ) + + @pytest.mark.parametrize("base", SENSOR_DELEGATE_BASES) + @pytest.mark.parametrize("state", [None, {}]) + def test_on_stream_event_out__without_state(self, base, state, stream, event): + # No state recorded for this sensor: it must still be called, with None. + delegate, [sensor] = self._delegate(base) + delegate.on_stream_event_out(TP1, 303, stream, event, state) + sensor.on_stream_event_out.assert_called_once_with( + TP1, 303, stream, event, None + ) + + @pytest.mark.parametrize("base", SENSOR_DELEGATE_BASES) + def test_direct_mutation_of_sensors_is_picked_up(self, base): + # The Cython version walks a list snapshot of the sensor set, so it + # has to notice a set that was mutated behind its back. + delegate, _ = self._delegate(base) + extra = Mock(name="extra", autospec=Sensor) + delegate._sensors.add(extra) + delegate.on_message_in(TP1, 3, None) + extra.on_message_in.assert_called_once_with(TP1, 3, None) diff --git a/tests/unit/transport/test_utils.py b/tests/unit/transport/test_utils.py index f7a2e1ab7..66bda8d2d 100644 --- a/tests/unit/transport/test_utils.py +++ b/tests/unit/transport/test_utils.py @@ -1,6 +1,20 @@ -from faust.transport.utils import DefaultSchedulingStrategy, TopicBuffer +import random + +import pytest + +from faust.transport.utils import ( + DefaultSchedulingStrategy, + TopicBuffer, + _py_records_iterator, + _records_iterator, +) from faust.types import TP +#: Both round-robin implementations. ``_records_iterator`` is the Cython one +#: whenever the extension could be built, and is otherwise the same object as +#: ``_py_records_iterator``. +RECORDS_ITERATOR_IMPLS = [_py_records_iterator, _records_iterator] + TP1 = TP("foo", 0) TP2 = TP("foo", 1) TP3 = TP("bar", 0) @@ -77,3 +91,118 @@ def test_next(self): (TP1, 3), (TP1, 4), ] + + +class Test_records_iterator: + def _index(self, records): + return DefaultSchedulingStrategy.map_from_records(records) + + @pytest.mark.parametrize("impl", RECORDS_ITERATOR_IMPLS) + def test_round_robin_over_topics_and_partitions(self, impl): + records = {TP1: BUF1, TP2: BUF2, TP3: BUF3, TP4: BUF4, TP5: BUF5} + + # Round-robin across topics, and across the partitions within each + # topic. "baz" only has one partition, so it drains early and the + # interleaving shifts once it and "bar" are exhausted. + assert list(impl(self._index(records))) == [ + (TP1, 0), + (TP3, 9), + (TP5, 14), + (TP2, 5), + (TP4, 11), + (TP5, 15), + (TP1, 1), + (TP3, 10), + (TP2, 6), + (TP4, 12), + (TP1, 2), + (TP4, 13), + (TP2, 7), + (TP1, 3), + (TP2, 8), + (TP1, 4), + ] + + @pytest.mark.parametrize("impl", RECORDS_ITERATOR_IMPLS) + def test_empty(self, impl): + assert list(impl(self._index({}))) == [] + + @pytest.mark.parametrize("impl", RECORDS_ITERATOR_IMPLS) + def test_empty_buffers(self, impl): + assert list(impl(self._index({TP1: [], TP3: []}))) == [] + + @pytest.mark.parametrize("impl", RECORDS_ITERATOR_IMPLS) + def test_drains_the_index_it_was_given(self, impl): + index = self._index({TP1: BUF1, TP3: BUF3}) + list(impl(index)) + assert not index + + @pytest.mark.parametrize("impl", RECORDS_ITERATOR_IMPLS) + def test_propagates_exceptions(self, impl): + def raising(): + yield (TP1, 1) + raise RuntimeError("buffer exploded") + + with pytest.raises(RuntimeError): + list(impl({"foo": raising()})) + + @pytest.mark.parametrize("impl", RECORDS_ITERATOR_IMPLS) + def test_topic_buffer_subclass(self, impl): + # A custom TopicBuffer must still be driven through next(), so that + # any overridden __iter__/__next__ is honoured. + class MyBuffer(TopicBuffer): + pass + + buffer = MyBuffer() + buffer.add(TP1, BUF1) + assert list(impl({"foo": buffer})) == [(TP1, i) for i in BUF1] + + def test_implementations_agree(self): + # Randomised differential test: the accelerated iterator must emit + # exactly the same sequence as the pure-Python one for any topology. + rng = random.Random(20220613) + for _ in range(200): + records = { + TP(f"t{topic}", partition): [ + f"t{topic}-{partition}-{i}" for i in range(rng.randint(0, 6)) + ] + for topic in range(rng.randint(0, 4)) + for partition in range(rng.randint(1, 4)) + } + assert list(_records_iterator(self._index(records))) == list( + _py_records_iterator(self._index(records)) + ) + + @pytest.mark.parametrize("impl", RECORDS_ITERATOR_IMPLS) + def test_partition_added_mid_iteration(self, impl): + # Both implementations re-read the buffer map on each pass. + buffer = TopicBuffer() + buffer.add(TP1, [1, 2]) + index = {"foo": buffer} + + it = impl(index) + buffer.add(TP2, [7, 8]) + assert list(it) == [(TP1, 1), (TP2, 7), (TP1, 2), (TP2, 8)] + + @pytest.mark.parametrize("impl", RECORDS_ITERATOR_IMPLS) + def test_topic_added_mid_iteration(self, impl): + buffer = TopicBuffer() + buffer.add(TP1, [1, 2]) + index = {"foo": buffer} + + it = impl(index) + late = TopicBuffer() + late.add(TP3, [9]) + index["bar"] = late + assert list(it) == [(TP1, 1), (TP3, 9), (TP1, 2)] + + @pytest.mark.parametrize("impl", RECORDS_ITERATOR_IMPLS) + def test_drains_the_buffer_map_too(self, impl): + # Exhausted partitions are popped from TopicBuffer._buffers. + buffer = TopicBuffer() + buffer.add(TP1, [1, 2]) + index = {"foo": buffer} + + list(impl(index)) + assert not index + assert not buffer._buffers diff --git a/tests/unit/utils/test_functional.py b/tests/unit/utils/test_functional.py index c6fa89b00..ec03d3b6c 100644 --- a/tests/unit/utils/test_functional.py +++ b/tests/unit/utils/test_functional.py @@ -1,21 +1,75 @@ import pytest -from faust.utils.functional import consecutive_numbers, translate +from faust.utils.functional import ( + _py_first_consecutive_run, + consecutive_numbers, + first_consecutive_run, + translate, +) + +#: Both implementations of ``first_consecutive_run``. ``first_consecutive_run`` +#: is the Cython one whenever the extension could be built, and is otherwise the +#: same object as ``_py_first_consecutive_run`` (in which case this just runs +#: the pure-Python one twice). +FIRST_CONSECUTIVE_RUN_IMPLS = [_py_first_consecutive_run, first_consecutive_run] + +RUN_CASES = [ + ([1, 2, 3, 4, 6, 7, 8], [1, 2, 3, 4]), + ([1, 4, 6, 8, 10], [1]), + ([1], [1]), + ([103, 104, 105, 106, 100000000000], [103, 104, 105, 106]), + # a run of numbers too large for a C long long must still work + ([2**80, 2**80 + 1, 2**80 + 3], [2**80, 2**80 + 1]), + # ... including when the overflow happens part-way through a run + ([1, 2, 2**80], [1, 2]), + # repeated numbers end a run, matching itertools.groupby + ([1, 1, 2], [1]), + ([0, 1, 2], [0, 1, 2]), + # descending numbers are not a run + ([5, 4, 3], [5]), +] @pytest.mark.parametrize( "numbers,expected", - [ - ([1, 2, 3, 4, 6, 7, 8], [1, 2, 3, 4]), - ([1, 4, 6, 8, 10], [1]), - ([1], [1]), - ([103, 104, 105, 106, 100000000000], [103, 104, 105, 106]), - ], + [(numbers, expected) for numbers, expected in RUN_CASES if numbers], ) def test_consecutive_numbers(numbers, expected): assert next(consecutive_numbers(numbers), None) == expected +@pytest.mark.parametrize("impl", FIRST_CONSECUTIVE_RUN_IMPLS) +@pytest.mark.parametrize("numbers,expected", RUN_CASES) +def test_first_consecutive_run(impl, numbers, expected): + assert impl(list(numbers)) == expected + + +@pytest.mark.parametrize("impl", FIRST_CONSECUTIVE_RUN_IMPLS) +def test_first_consecutive_run__empty(impl): + assert impl([]) == [] + + +@pytest.mark.parametrize("impl", FIRST_CONSECUTIVE_RUN_IMPLS) +def test_first_consecutive_run__accepts_any_iterable(impl): + assert impl(iter([1, 2, 3, 5])) == [1, 2, 3] + assert impl(range(4)) == [0, 1, 2, 3] + + +@pytest.mark.parametrize("impl", FIRST_CONSECUTIVE_RUN_IMPLS) +def test_first_consecutive_run__does_not_mutate_argument(impl): + numbers = [1, 2, 5] + assert impl(numbers) == [1, 2] + assert numbers == [1, 2, 5] + + +@pytest.mark.parametrize("impl", FIRST_CONSECUTIVE_RUN_IMPLS) +@pytest.mark.parametrize("numbers,expected", RUN_CASES) +def test_first_consecutive_run__matches_consecutive_numbers(impl, numbers, expected): + # the helper must return exactly what taking the first group of + # consecutive_numbers() would have returned. + assert impl(list(numbers)) == next(consecutive_numbers(numbers), []) + + @pytest.mark.parametrize( "table,s,expected", [ @@ -28,3 +82,42 @@ def test_consecutive_numbers(numbers, expected): ) def test_translate(table, s, expected): assert translate(table, s) == expected + + +@pytest.mark.parametrize("impl", FIRST_CONSECUTIVE_RUN_IMPLS) +def test_first_consecutive_run__is_lazy(impl): + # Must stop consuming as soon as the run is broken, leaving the rest of + # the iterator for the caller (same as next(consecutive_numbers(...))). + it = iter([1, 2, 5, 6, 7]) + assert impl(it) == [1, 2] + assert list(it) == [6, 7] + + +@pytest.mark.parametrize("impl", FIRST_CONSECUTIVE_RUN_IMPLS) +def test_first_consecutive_run__does_not_over_consume(impl): + # An iterable that blows up after the run has ended must not be reached. + def numbers(): + yield 1 + yield 2 + yield 99 + raise AssertionError("consumed past the end of the run") + + assert impl(numbers()) == [1, 2] + + +@pytest.mark.parametrize("impl", FIRST_CONSECUTIVE_RUN_IMPLS) +def test_first_consecutive_run__list_resized_while_scanning(impl): + # __sub__ can run arbitrary Python code, including code that shrinks the + # list being scanned -- which must not read past the end of it. + class Shrinks: + def __init__(self, container=None): + self.container = container + + def __sub__(self, other): + if self.container is not None: + del self.container[2:] + return 1 + + numbers = [] + numbers.extend([Shrinks(), Shrinks(numbers)] + [Shrinks() for _ in range(20)]) + assert len(impl(numbers)) <= 2 From e2c92b8c03151f6d373c1d315c49693b0e7e5292 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 4 Aug 2026 12:51:44 +0000 Subject: [PATCH 3/7] Port the batch-shaped offset scan to Rust and measure it docs/proposals/rust-acceleration.md said the recommendation would change if "a batch-shaped hot path appears in Faust's own code". One has: first_consecutive_run, which Consumer._new_offset calls once per assigned partition per commit to scan the sorted list of acked offsets. At a busy partition that is one boundary crossing followed by 600k iterations, and it blocks the event loop while it runs. So it was ported to Rust and measured. Rust lost, and the reason sharpens the whole evaluation. At 100k offsets, against the Cython that ships today (549us): python 2775us 0.20x rust, idiomatic PyO3, abi3 1793us 0.31x rust, idiomatic PyO3, no abi3 1184us 0.46x rust, raw pyo3::ffi, abi3 831us 0.66x rust, raw pyo3::ffi + CPython macros 648us 0.85x rust, same scan over a native Vec 83us 6.8x Rust at its best is still 15% slower, and that best case is unsafe raw-pointer code that has given up both memory safety and abi3 -- the two reasons to prefer Rust in the first place. The last row explains the rest: the loop is batch-shaped, but every element is a PyObject, so each iteration is three C-API calls that Rust makes at the same price Cython does. Handed the same data natively, the identical scan is 6.8x faster than Cython. Being batch-shaped was therefore the wrong test. Section 6's trigger is rewritten: what matters is whether the per-element work is native or Python-object-bound, and the practical check before porting anything is to count the C-API calls per element. Two smaller corrections fall out of it: - abi3 and performance are in direct tension. The limited API does not expose PyList_GET_ITEM, so the fastest variant cannot be compiled in an abi3 build: best-with-abi3 is 0.66x, best-without is 0.85x. - Section 3.1's HoppingWindow port is now on-branch and re-measured. It does better than the original off-branch one (mostly lto plus codegen-units = 1): ranges is 2.26x rather than 1.71x, and the three scalar methods are a wash rather than a 0.83-0.87x loss. The conclusion is unchanged. The crate is not wired into setup.py, for the reason the document already gives: PEP 518 has no conditional build-requires, so wiring it would make setuptools-rust a mandatory build dependency for everyone. It builds via extra/tools/build_rust_accel.sh instead, and faust/_rust/ has no __init__.py so find_packages() cannot pull it into the wheel. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_019oX4oGCQGgvPJHabjwfaBk --- .gitignore | 1 + docs/proposals/rust-acceleration.md | 183 +++++++++++++--- extra/tools/bench_accel_offsets.py | 122 +++++++++++ extra/tools/build_rust_accel.sh | 37 ++++ faust/_rust/.gitignore | 2 + faust/_rust/Cargo.lock | 132 ++++++++++++ faust/_rust/Cargo.toml | 29 +++ faust/_rust/src/lib.rs | 323 ++++++++++++++++++++++++++++ 8 files changed, 801 insertions(+), 28 deletions(-) create mode 100644 extra/tools/bench_accel_offsets.py create mode 100755 extra/tools/build_rust_accel.sh create mode 100644 faust/_rust/.gitignore create mode 100644 faust/_rust/Cargo.lock create mode 100644 faust/_rust/Cargo.toml create mode 100644 faust/_rust/src/lib.rs diff --git a/.gitignore b/.gitignore index be040cade..2f1d3b601 100644 --- a/.gitignore +++ b/.gitignore @@ -52,3 +52,4 @@ faust/sensors/_cython/*.c # coverage .coverage coverage.xml +faust/_rust/target/ diff --git a/docs/proposals/rust-acceleration.md b/docs/proposals/rust-acceleration.md index 01cac5116..0ec5a7faf 100644 --- a/docs/proposals/rust-acceleration.md +++ b/docs/proposals/rust-acceleration.md @@ -7,14 +7,24 @@ alongside (or eventually instead of) its Cython extensions. **Recommendation in one line:** do not add a Rust build axis to Faust today. The prototype works and the build integration is sound, but on the code Faust -actually accelerates, Rust is between 1.9x faster and 15% *slower* than the -Cython we already ship, and it would roughly double the test matrix and add a -second native toolchain for that. Prefer keeping Cython, and revisit only if a -genuinely batch-shaped hot path appears (see -[What would change this answer](#what-would-change-this-answer)). +actually accelerates, Rust lands between 2.3x faster and 15% *slower* than the +Cython we already ship — and it is only faster on one method of one class, +while being slower on the single busiest loop Faust owns. It would roughly +double the test matrix and add a second native toolchain for that. Prefer keeping Cython, and revisit only if a +hot path appears whose per-element work is *native* rather than +Python-object-bound (see +[What would change this answer](#6-what-would-change-this-answer)). + +> **Update.** The batch-shaped hot path this document was waiting for has +> since landed — `first_consecutive_run`, which scans up to hundreds of +> thousands of acked offsets per commit. It was ported to Rust and measured. +> **Rust lost anyway**, and the reason turned out to sharpen the whole +> evaluation: being batch-shaped is not sufficient. See +> [§3.3](#33-first_consecutive_run-the-batch-shaped-path-that-arrived), which +> also corrects the trigger in §6 that this update fires. Everything below was measured or built, not estimated; see -[Reproducing](#reproducing). +[Reproducing](#7-reproducing). --- @@ -132,16 +142,32 @@ extensions built with the project's standard `-O2`. `min` of 5 runs × scalar methods at 0.87–0.91x; run-to-run spread is a few percent, the sign is stable.) +**Correction, from re-running this against the on-branch port.** The `HoppingWindow` +port now lives in `faust/_rust/src/lib.rs` (it was off-branch when the table +above was written), so anyone can re-run it — and it measures better than the +original did. On the same machine as §3.3: `ranges` **2.26x faster** than +Cython (773 → 341 ns), `current` 1.02x, `stale` 0.98x, `earliest` 0.92x. +So the three scalar methods are a **wash**, not the 0.83–0.87x loss recorded +above; the difference is mostly `lto = true` plus `codegen-units = 1`, which +the original crate did not set. `abi3` makes no measurable difference here — +unlike §3.3, this code indexes no lists, so it never touches the macros the +limited API hides. + +The conclusion is unchanged, and if anything the corrected numbers make it +cleaner: **Rust wins where there is work to do per call and ties where there +is not.** Nothing here justifies a second toolchain. + The pattern is the important part, and it is not about Rust being slow: * **The win is proportional to work done per call.** `ranges` builds a list of - ~7 tuples and Rust wins 1.7x. The three methods that do a couple of - floating-point operations and return one tuple all *lose* to Cython. + ~7 tuples and Rust wins outright. The three methods that do a couple of + floating-point operations and return one tuple land on top of Cython. * **Because the floor is the call boundary, not the language.** A bare - attribute read costs **60 ns through Cython and 79 ns through PyO3** — - PyO3's argument parsing, `Bound` handling and error plumbing are simply - thicker than `cdef class` access. Any call whose body is cheaper than ~30 ns - of that difference is a guaranteed loss, whatever the language. + attribute read costs **43 ns through Cython and 52 ns through PyO3** on the + re-run (60/79 ns originally) — PyO3's argument parsing, `Bound` handling and + error plumbing are simply thicker than `cdef class` access. Any call whose + body is cheaper than that difference is a guaranteed loss, whatever the + language. ### 3.2 Build cost @@ -156,6 +182,64 @@ almost entirely compiling the PyO3 macro stack (`syn`, `quote`, `proc-macro2`, `pyo3-macros`), and it is a fixed cost that does not grow much as Faust's own Rust grows. It is paid per CI leg, not once. +### 3.3 `first_consecutive_run`: the batch-shaped path that arrived + +§6 said the answer would change if "a batch-shaped hot path appears in Faust's +own code — something that crosses the boundary once and then does thousands of +operations". One has: `faust/utils/_cython/functional.pyx`'s +`first_consecutive_run`, called once per assigned partition per commit from +`Consumer._new_offset`, scanning the sorted list of acked offsets. At a busy +partition that list holds hundreds of thousands of entries, so a single call +does one boundary crossing and then 600 000 iterations. It is about as +batch-shaped as Faust gets, and it blocks the event loop while it runs. + +It was ported to Rust four ways, each one giving up more safety than the last. +100 000 offsets, all consecutive (the worst case — the whole list is scanned), +same machine and method as §3.1: + +| implementation | 100k offsets | vs cython | +| --- | ---: | ---: | +| pure Python | 2775 µs | 0.20x | +| **Cython (what ships today)** | **549 µs** | **1.00x** | +| Rust, idiomatic PyO3, `abi3` | 1793 µs | 0.31x | +| Rust, idiomatic PyO3, no `abi3` | 1184 µs | 0.46x | +| Rust, raw `pyo3::ffi`, `abi3` | 831 µs | 0.66x | +| Rust, raw `pyo3::ffi` + CPython macros, no `abi3` | 648 µs | 0.85x | +| *Rust, same scan over a native `Vec`* | *83 µs* | *6.8x* | + +Ratios hold within a few percent across 1k / 10k / 100k / 600k offsets, so this +is not a fixed-overhead artefact — it is the per-element cost. + +**Rust at its absolute best is still 15% slower than the Cython already in the +tree.** That best case is `unsafe` raw-pointer code calling `PyList_GET_ITEM` +and `PyLong_AsLongLongAndOverflow` directly — which is to say, it is C with +Rust syntax, and it has given up both memory safety and `abi3`, the two things +that were the reasons to prefer Rust in the first place. + +The last row is the one that explains all the others, and it is the real +finding here: + +> The loop is batch-shaped, but every element in it is a `PyObject`. Each +> iteration is `PyList_GET_ITEM` + `PyLong_AsLongLongAndOverflow` + +> `PyList_Append` — three C-API calls that Rust has to make just as Cython +> does, at the same price. Handed the *same data* as a native `Vec`, the +> identical scan is **6.8x faster than Cython**. The bottleneck was never the +> language; it is that the data is Python objects from end to end. + +Two corollaries worth carrying forward: + +1. **"Batch-shaped" was the wrong test.** Crossing the boundary once is + necessary but nowhere near sufficient. What matters is whether the + per-element work is native or Python-object-bound. A loop that touches a + `PyObject` per iteration has already lost, no matter how long it runs; + §6.1 is rewritten accordingly. +2. **`abi3` and performance are in direct tension.** The limited API does not + expose `PyList_GET_ITEM`/`PyList_GET_SIZE`, so the fastest variant *cannot + be compiled* in an `abi3` build — best-with-`abi3` is 0.66x, best-without + is 0.85x. §5's "the abi3 upside does not materialise" is stronger than it + was written: even where abi3 is available, taking it costs ~22% on this + kind of code. + ## 4. Which Faust code could actually go to Rust | Candidate | Shape | Verdict | @@ -168,11 +252,18 @@ as Faust's own Rust grows. It is paid per CI leg, not once. | State store | Batch-shaped, big | **Already solved** by `faust[rocksdict]`, which is PyO3. | | Codec chains (`faust/serializers/codecs.py`) | Thin wrappers over `json`/`pickle`/`base64` | No meaningful compute of our own to move. | | Model field coercion (`faust/models/`) | Plausible on paper | Deeply coupled to `typing` introspection and user-supplied Python callables; would cross the boundary constantly. Not evaluated further. | +| `faust/utils/_cython/functional.pyx` (`first_consecutive_run`) | Batch-shaped, but every element is a `PyObject` | **Ported and measured — Rust loses.** 0.85x Cython at its unsafe, non-`abi3` best; 0.66x with `abi3`. See §3.3. | +| `faust/transport/_cython/scheduler.pyx` (`records_iterator`) | Round-robin cursor over Python lists, once per record | **Poor fit.** ~50 ns/record in Cython, and the body is `PyIter_Next` + tuple building — Python-object work end to end, with a call boundary per record on top. | +| `faust/sensors/_cython/base.pyx` (`SensorDelegateBase`) | Iterates a set and calls back into Python 4x per message | **Poor fit**, and the worst of the three: the entire body *is* calls into Python, which is precisely the 79 ns-per-crossing case from §3.1. | The summary of that table is the crux of the evaluation: **every part of Faust -whose shape suits Rust is already served by someone else's Rust**, and the -three things Faust accelerates itself are all latency-bound call-boundary code, -which is the one workload where Cython beats PyO3. +whose shape suits Rust is already served by someone else's Rust**, and +everything Faust accelerates itself is either latency-bound call-boundary code +or a loop over Python objects — the two workloads where Cython beats PyO3. + +The three accelerators added since this document was first written did not +change that. Two are call-boundary code, and the third (§3.3) looked like the +exception and turned out not to be. ## 5. What it would cost @@ -221,12 +312,26 @@ real cost. Concrete triggers, in rough order of likelihood: -1. **A batch-shaped hot path appears in Faust's own code** — something that - crosses the boundary once and then does thousands of operations - (windowed-table range scans over many keys, bulk changelog - reconstruction on recovery, a partition-wide sort/merge). §3.1 shows the - win scales with work-per-call; `ranges`, the most batch-like of four - methods, is the only one that wins. +1. ~~**A batch-shaped hot path appears in Faust's own code**~~ — **tested and + wrong; superseded by 1a.** One did appear (`first_consecutive_run`, + §3.3) and Rust lost to Cython anyway. Crossing the boundary once is + necessary but not sufficient, because the per-element cost dominates and + that cost is the same in both languages when the elements are `PyObject`s. + +1a. **A hot path appears whose per-element work is *native*** — where the data + can be converted to a native representation once (or, better, already + *is* one) and then scanned without touching a `PyObject` per iteration. + §3.3 measured the gap this makes: the same scan is 0.85x Cython over a + Python list and **6.8x** over a `Vec`. + The practical test before porting anything: *count the C-API calls per + element.* If it is more than zero, expect to lose. + Candidates that might genuinely qualify — none of them evaluated yet — + are ones where Faust could own the buffer end to end: offset bookkeeping + held as an `i64` array instead of `list[int]`, changelog reconstruction + over raw bytes on recovery, or a windowed-table range scan that returns + an aggregate rather than a list of Python tuples. Note that the first of + those is a data-structure change to `Consumer._acked`, not a language + choice, and would speed up the Cython version too. 2. **Profiling shows the Cython extensions are actually material.** Nobody has published what fraction of Faust's per-message cost lives in `streams.pyx` + `conductor.pyx`. If it is small, the entire accelerator @@ -240,19 +345,41 @@ Concrete triggers, in rough order of likelihood: time, or the `cp3*`/free-threading friction already visible in `pyproject.toml`'s skip comment. -If (1) or (3) lands, the §2 prototype is the implementation: it is about 40 +If (1a) or (3) lands, the §2 prototype is the implementation: it is about 40 lines of build wiring, it degrades correctly, and it has been shown to work. ## 7. Reproducing +The Rust crate used for §3.3 now lives in `faust/_rust/`. It is deliberately +**not** wired into `setup.py`: §2's integration point 2 still applies — PEP 518 +has no conditional `build-requires`, so wiring it up would make +`setuptools-rust` a mandatory build dependency for everyone, which is a real +cost to impose for a prototype this document recommends against shipping. The +build script below stands in for that wiring. + ```bash python -m venv .venv && . .venv/bin/activate pip install -e . # builds the Cython extensions -python extra/tools/bench_accel_windows.py # python vs cython (vs rust, if built) + +python extra/tools/bench_accel_windows.py # §3.1 +python extra/tools/bench_accel_offsets.py # §3.3 + +# optional: add the rust columns to both +extra/tools/build_rust_accel.sh # abi3, as a real PR would ship +extra/tools/build_rust_accel.sh --no-abi3 # adds the macro-based variants ``` -The benchmark skips any implementation it cannot import, so it is useful on a -pure-Python install too. To reproduce the `rust` column, apply the §2 wiring, -port `faust/_cython/windows.pyx`'s `HoppingWindow` to -`faust/_rust/src/lib.rs`, and build with `USE_RUST=1 pip install -e .` — the -benchmark picks up `faust._rust._accel` automatically. +Both benchmarks skip any implementation they cannot import, so they are useful +on a pure-Python install too, and `bench_accel_offsets.py` refuses to report +timings for implementations that disagree on its correctness cases. + +Note that the `rust/macro` row only exists in a `--no-abi3` build: the variants +are gated behind `#[cfg(not(feature = "abi3"))]` because the limited API does +not expose `PyList_GET_ITEM`. That gate is the §3.3 corollary in code form. + +`faust/_rust/` deliberately has **no `__init__.py`**. It is imported as a PEP +420 namespace package, which is what keeps `find_packages()` from picking it +up and shipping a dead directory inside the wheel — the crate stays entirely +outside the distribution, with no `exclude` entry needed in `setup.py`. If a +real Rust PR is ever written, that changes: it would add the `__init__.py` +along with everything in §1's six-row table. diff --git a/extra/tools/bench_accel_offsets.py b/extra/tools/bench_accel_offsets.py new file mode 100644 index 000000000..6e67770f8 --- /dev/null +++ b/extra/tools/bench_accel_offsets.py @@ -0,0 +1,122 @@ +#!/usr/bin/env python +"""Compare the available ``first_consecutive_run`` implementations. + +``Consumer._new_offset`` scans the sorted list of acked offsets for a +partition on every commit, which is the most batch-shaped hot path Faust has +in its own code -- one call, then up to hundreds of thousands of iterations. +``docs/proposals/rust-acceleration.md`` names exactly that shape as the +trigger for revisiting Rust, so this script measures it. + +Usage:: + + pip install -e . # builds faust._cython.functional + extra/tools/build_rust_accel.sh --no-abi3 # optional, adds the rust columns + python extra/tools/bench_accel_offsets.py + +Any implementation that is not importable is skipped, so this is useful on a +pure-Python install too. +""" + +import timeit +from typing import Any, Callable, List, Tuple + +from faust.utils.functional import _py_first_consecutive_run + +SIZES = (1_000, 10_000, 100_000, 600_000) +MIN_TIME = 0.25 +REPEAT = 5 + +Impl = Tuple[str, Callable[[List[int]], Any]] + + +def _implementations() -> List[Impl]: + impls: List[Impl] = [("python", _py_first_consecutive_run)] + try: + from faust.utils._cython.functional import first_consecutive_run + except ImportError: + pass + else: + impls.append(("cython", first_consecutive_run)) + + # Not built by default; see docs/proposals/rust-acceleration.md. + try: + from faust._rust import _accel + except ImportError: + return impls + # Ordered worst to best, which is also least to most unsafe. + for name, attr in ( + ("rust/pyo3", "first_consecutive_run"), + ("rust/ffi", "first_consecutive_run_ffi"), + ("rust/macro", "first_consecutive_run_macro"), + ): + fn = getattr(_accel, attr, None) + if fn is not None: + impls.append((name, fn)) + return impls + + +def _check(impls: List[Impl]) -> None: + """Fail loudly rather than benchmark implementations that disagree.""" + cases = [ + [1, 2, 3, 4, 6, 7, 8], + [1, 4, 6, 8, 10], + [1], + [], + [0, 1, 2], + [1, 1, 2], + list(range(100)), + ] + reference_name, reference = impls[0] + for case in cases: + want = reference(list(case)) + for name, fn in impls[1:]: + got = fn(list(case)) + if got != want: + raise SystemExit( + f"{name} disagrees with {reference_name} on {case!r}: " + f"{got!r} != {want!r}" + ) + print(f"{len(impls)} implementations agree on {len(cases)} cases\n") + + +def _ns_per_call(fn: Callable[[List[int]], Any], data: List[int]) -> float: + fn(data) + number = 1 + while timeit.timeit(lambda: fn(data), number=number) < MIN_TIME: + number *= 4 + best = min(timeit.timeit(lambda: fn(data), number=number) for _ in range(REPEAT)) + return best / number * 1e9 + + +def main() -> int: + impls = _implementations() + _check(impls) + + width = max(len(name) for name, _ in impls) + 2 + header = f"{'offsets':>9}" + "".join(f"{name:>{width + 4}}" for name, _ in impls) + print(header) + print("-" * len(header)) + + baselines = {} + for size in SIZES: + data = list(range(size)) + row = f"{size:>9}" + for name, fn in impls: + micros = _ns_per_call(fn, data) / 1000.0 + baselines.setdefault(name, []).append(micros) + row += f"{micros:>{width}.1f} us" + print(row) + + if "cython" in baselines: + print("\nrelative to cython (higher is faster):") + for name, values in baselines.items(): + ratios = [ + cython / value for cython, value in zip(baselines["cython"], values) + ] + span = f"{min(ratios):.2f}x - {max(ratios):.2f}x" + print(f" {name:>12}: {span}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/extra/tools/build_rust_accel.sh b/extra/tools/build_rust_accel.sh new file mode 100755 index 000000000..aed1264e2 --- /dev/null +++ b/extra/tools/build_rust_accel.sh @@ -0,0 +1,37 @@ +#!/bin/sh +# Build the evaluation-only Rust accelerator in faust/_rust/. +# +# This crate is deliberately NOT wired into setup.py: see +# docs/proposals/rust-acceleration.md for why (PEP 518 has no conditional +# build-requires, so wiring it up would make setuptools-rust a mandatory +# build dependency for everyone). This script exists so the numbers in that +# document can be re-checked without changing Faust's build. +# +# Usage: +# extra/tools/build_rust_accel.sh # abi3 (what a real PR would ship) +# extra/tools/build_rust_accel.sh --no-abi3 # adds the macro-based variants +# +# Then: +# python extra/tools/bench_accel_offsets.py +set -e + +CRATE_DIR="$(dirname "$0")/../../faust/_rust" +cd "$CRATE_DIR" + +# Remove any previous build FIRST: if the compile below fails, a stale module +# left in place would be picked up by the benchmark and silently reported as +# the build you asked for. +rm -f _accel.so _accel.abi3.so + +if [ "$1" = "--no-abi3" ]; then + echo "--- building without abi3 (enables the macro-based variants) ---" + cargo build --release --no-default-features + OUT="_accel.so" +else + echo "--- building with abi3-py310 ---" + cargo build --release + OUT="_accel.abi3.so" +fi + +cp target/release/lib_accel.so "$OUT" +echo "built $CRATE_DIR/$OUT" diff --git a/faust/_rust/.gitignore b/faust/_rust/.gitignore new file mode 100644 index 000000000..b857562aa --- /dev/null +++ b/faust/_rust/.gitignore @@ -0,0 +1,2 @@ +target/ +*.so diff --git a/faust/_rust/Cargo.lock b/faust/_rust/Cargo.lock new file mode 100644 index 000000000..6c23f2607 --- /dev/null +++ b/faust/_rust/Cargo.lock @@ -0,0 +1,132 @@ +# This file is automatically @generated by Cargo. +# It is not intended for manual editing. +version = 4 + +[[package]] +name = "faust-accel" +version = "0.1.0" +dependencies = [ + "pyo3", +] + +[[package]] +name = "heck" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" + +[[package]] +name = "libc" +version = "0.2.189" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2" + +[[package]] +name = "once_cell" +version = "1.21.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" + +[[package]] +name = "portable-atomic" +version = "1.14.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3d20d5497ef88037a52ff98267d066e7f11fcc5e99bbfbd58a42336193aacec3" + +[[package]] +name = "proc-macro2" +version = "1.0.107" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "pyo3" +version = "0.29.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc153f5fd745cc038b5eed86622125969f8a39834a57bc96beaaf2512b1da729" +dependencies = [ + "libc", + "once_cell", + "portable-atomic", + "pyo3-build-config", + "pyo3-ffi", + "pyo3-macros", +] + +[[package]] +name = "pyo3-build-config" +version = "0.29.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bb77d9aa6d647507b55c69ee714d266d84c526c78ff0bc6dd8757f58591e64d1" +dependencies = [ + "target-lexicon", +] + +[[package]] +name = "pyo3-ffi" +version = "0.29.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3160087aa5733bce7d9a729f8c55f85a2a53b74742a09613178eeebff0722253" +dependencies = [ + "libc", + "pyo3-build-config", +] + +[[package]] +name = "pyo3-macros" +version = "0.29.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91f9d455db760a9a0b0ddeaac25f1390b8a36ba73dfbda9f127cac6fc340d4d5" +dependencies = [ + "proc-macro2", + "pyo3-macros-backend", + "quote", + "syn", +] + +[[package]] +name = "pyo3-macros-backend" +version = "0.29.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e343bcec300ff262f5806a33a4e51b6d097a8a46435f512fcb83e95592581625" +dependencies = [ + "heck", + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "quote" +version = "1.0.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "syn" +version = "2.0.119" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "872831b642d1a07999a962a351ed35b955ea2cfc8f3862091e2a240a84f17297" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "target-lexicon" +version = "0.13.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "adb6935a6f5c20170eeceb1a3835a49e12e19d792f6dd344ccc76a985ca5a6ca" + +[[package]] +name = "unicode-ident" +version = "1.0.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" diff --git a/faust/_rust/Cargo.toml b/faust/_rust/Cargo.toml new file mode 100644 index 000000000..3f2505b0c --- /dev/null +++ b/faust/_rust/Cargo.toml @@ -0,0 +1,29 @@ +[package] +name = "faust-accel" +version = "0.1.0" +edition = "2021" +publish = false + +# Evaluation only -- see docs/proposals/rust-acceleration.md. +# Nothing in Faust's build system references this crate; build it with +# extra/tools/build_rust_accel.sh when you want to re-check the numbers. + +[lib] +name = "_accel" +crate-type = ["cdylib"] + +[features] +# abi3 is what a real PR would ship: one build covers CPython 3.10-3.14. +# Turning it off is what the proposal's measurements need, because the +# limited API hides the CPython macros the Cython extension compiles down to. +default = ["abi3"] +abi3 = ["pyo3/abi3-py310"] + +[dependencies.pyo3] +version = "0.29" +features = ["extension-module"] + +[profile.release] +opt-level = 3 +lto = true +codegen-units = 1 diff --git a/faust/_rust/src/lib.rs b/faust/_rust/src/lib.rs new file mode 100644 index 000000000..4491e2de8 --- /dev/null +++ b/faust/_rust/src/lib.rs @@ -0,0 +1,323 @@ +//! Rust ports of Faust's Cython accelerators, for the evaluation in +//! `docs/proposals/rust-acceleration.md`. +//! +//! Not built by default. See that document for the `USE_RUST` build wiring +//! and for what the measurements mean. + +use pyo3::prelude::*; +use pyo3::types::{PyInt, PyList, PyTuple}; + +/// Port of `faust._cython.functional.first_consecutive_run`. +/// +/// Kept deliberately faithful to the Cython version so the comparison is +/// like-for-like: the same two-tier structure (an `i64` fast path for exact +/// ints, a Python-arithmetic slow path for everything else), and the same +/// rule that a run continues only while each number is exactly one greater +/// than the one before it. +#[pyfunction] +fn first_consecutive_run<'py>(numbers: &Bound<'py, PyAny>) -> PyResult> { + match numbers.cast::() { + Ok(seq) => run_from_list(seq), + Err(_) => run_from_iterable(numbers), + } +} + +fn run_from_list<'py>(seq: &Bound<'py, PyList>) -> PyResult> { + let py = seq.py(); + if seq.len() == 0 { + return Ok(PyList::empty(py)); + } + + let mut prev = seq.get_item(0)?; + let out = PyList::empty(py); + out.append(&prev)?; + let mut i: usize = 1; + + // Fast path: exact ints that fit in an i64. `is_exact_instance_of` + // matches Cython's PyLong_CheckExact, so bool (an int subclass) falls + // through to the slow path in both implementations. + if prev.is_exact_instance_of::() { + if let Ok(mut c_prev) = prev.extract::() { + // The length is re-read every iteration: nothing in this loop can + // run Python code, but the list is the caller's and staying in + // step with the Cython version costs nothing measurable. + while i < seq.len() { + let cur = seq.get_item(i)?; + if !cur.is_exact_instance_of::() { + break; + } + let c_cur = match cur.extract::() { + Ok(value) => value, + Err(_) => break, + }; + if c_prev.checked_add(1) != Some(c_cur) { + break; + } + out.append(&cur)?; + c_prev = c_cur; + i += 1; + } + if i >= seq.len() { + return Ok(out); + } + prev = seq.get_item(i - 1)?; + } + } + + // Slow path: arbitrary objects supporting `-` and comparison to 1. + while i < seq.len() { + let cur = seq.get_item(i)?; + if !cur.sub(&prev)?.eq(1i64)? { + break; + } + out.append(&cur)?; + prev = cur; + i += 1; + } + Ok(out) +} + +fn run_from_iterable<'py>(numbers: &Bound<'py, PyAny>) -> PyResult> { + let py = numbers.py(); + let mut it = numbers.try_iter()?; + let out = PyList::empty(py); + + let mut prev = match it.next() { + None => return Ok(out), + Some(first) => first?, + }; + out.append(&prev)?; + + // Stops consuming as soon as the run is broken, so a shared iterator is + // left exactly where the Python and Cython versions leave it. + for cur in it { + let cur = cur?; + if !cur.sub(&prev)?.eq(1i64)? { + break; + } + out.append(&cur)?; + prev = cur; + } + Ok(out) +} + +/// Length of the first consecutive run, without materialising it. +/// +/// Not a drop-in for anything Faust calls -- it exists to separate the two +/// costs the benchmark keeps conflating: scanning the input, and building the +/// Python list of results. See the proposal for why that distinction is the +/// whole answer. +#[pyfunction] +fn first_consecutive_run_length(seq: &Bound<'_, PyList>) -> PyResult { + if seq.len() == 0 { + return Ok(0); + } + let first = seq.get_item(0)?; + let mut c_prev = match first.extract::() { + Ok(value) => value, + Err(_) => return Ok(1), + }; + let mut count: usize = 1; + let mut i: usize = 1; + while i < seq.len() { + let cur = seq.get_item(i)?; + let c_cur = match cur.extract::() { + Ok(value) => value, + Err(_) => break, + }; + if c_prev.checked_add(1) != Some(c_cur) { + break; + } + count += 1; + c_prev = c_cur; + i += 1; + } + Ok(count) +} + +/// Port of `faust._cython.windows.HoppingWindow`. +/// +/// Only the float paths are implemented: enough to reproduce the numbers in +/// the proposal, not enough to stand in for the real class. +#[pyclass] +struct HoppingWindow { + #[pyo3(get)] + size: f64, + #[pyo3(get)] + step: f64, + #[pyo3(get)] + expires: f64, +} + +impl HoppingWindow { + fn start_initial_range(&self, timestamp: f64) -> f64 { + let rem = (timestamp / self.step).floor() as i64; + (rem as f64) * self.step - self.size + self.step + } + + fn current_start(&self, timestamp: f64, start: f64) -> f64 { + let m = ((timestamp - start) / self.step).floor(); + start + (self.step * m) + } +} + +#[pymethods] +impl HoppingWindow { + #[new] + #[pyo3(signature = (size, step, expires = 0.0))] + fn new(size: f64, step: f64, expires: f64) -> Self { + HoppingWindow { + size, + step, + expires, + } + } + + fn ranges<'py>(&self, py: Python<'py>, timestamp: f64) -> PyResult> { + let out = PyList::empty(py); + let mut start = self.start_initial_range(timestamp) as i64; + let stop = timestamp as i64; + let step = self.step as i64; + while start <= stop { + let begin = start as f64; + out.append(PyTuple::new(py, [begin, begin + self.size - 0.1])?)?; + start += step; + } + Ok(out) + } + + fn current<'py>(&self, py: Python<'py>, timestamp: f64) -> PyResult> { + let initial = self.start_initial_range(timestamp); + let start = self.current_start(timestamp, initial); + PyTuple::new(py, [start, start + self.size - 0.1]) + } + + fn earliest<'py>(&self, py: Python<'py>, timestamp: f64) -> PyResult> { + let start = self.start_initial_range(timestamp); + PyTuple::new(py, [start, start + self.size - 0.1]) + } + + fn delta<'py>(&self, py: Python<'py>, timestamp: f64, d: f64) -> PyResult> { + self.current(py, timestamp - d) + } + + fn stale(&self, timestamp: f64, latest_timestamp: f64) -> bool { + if self.expires == 0.0 { + return false; + } + let ts = latest_timestamp - self.expires; + let initial = self.start_initial_range(ts); + timestamp <= self.current_start(ts, initial) + } +} + +/// Same as `_ffi`, but using the CPython macros that the limited API hides. +/// Only compiles without `abi3`. +#[cfg(not(feature = "abi3"))] +#[pyfunction] +fn first_consecutive_run_macro<'py>(seq: &Bound<'py, PyList>) -> PyResult> { + use pyo3::ffi; + let py = seq.py(); + let out = PyList::empty(py); + let raw = seq.as_ptr(); + unsafe { + let n = ffi::PyList_GET_SIZE(raw); + if n == 0 { + return Ok(out); + } + let mut item = ffi::PyList_GET_ITEM(raw, 0); + ffi::PyList_Append(out.as_ptr(), item); + let mut overflow: std::os::raw::c_int = 0; + let mut c_prev = ffi::PyLong_AsLongLongAndOverflow(item, &mut overflow); + if overflow != 0 { + return Ok(out); + } + let mut i: ffi::Py_ssize_t = 1; + while i < n { + item = ffi::PyList_GET_ITEM(raw, i); + let c_cur = ffi::PyLong_AsLongLongAndOverflow(item, &mut overflow); + if overflow != 0 || c_cur <= c_prev || c_cur - 1 != c_prev { + break; + } + ffi::PyList_Append(out.as_ptr(), item); + c_prev = c_cur; + i += 1; + } + } + Ok(out) +} + +/// The same scan over data that is already native. +/// +/// Builds `0..n` as a `Vec` and finds the first consecutive run in it. +/// Nothing here touches a Python object, so this is the ceiling: what the +/// scan costs once the per-element Python object work is removed. +#[pyfunction] +fn scan_native(n: usize) -> usize { + let data: Vec = (0..n as i64).collect(); + if data.is_empty() { + return 0; + } + let mut count = 1usize; + let mut prev = data[0]; + for &cur in &data[1..] { + if cur != prev + 1 { + break; + } + count += 1; + prev = cur; + } + count +} + +#[pymodule] +fn _accel(m: &Bound<'_, PyModule>) -> PyResult<()> { + m.add_function(wrap_pyfunction!(first_consecutive_run, m)?)?; + m.add_function(wrap_pyfunction!(first_consecutive_run_length, m)?)?; + m.add_function(wrap_pyfunction!(first_consecutive_run_ffi, m)?)?; + #[cfg(not(feature = "abi3"))] + m.add_function(wrap_pyfunction!(first_consecutive_run_macro, m)?)?; + m.add_function(wrap_pyfunction!(scan_native, m)?)?; + m.add_class::()?; + Ok(()) +} + +/// Raw-FFI variant of `first_consecutive_run`, list fast path only. +/// +/// The idiomatic version above pays for a bounds-checked `get_item` that +/// returns an owned `Bound` (a refcount round-trip per element) and for +/// PyO3's `extract` machinery. This one calls the same C API the Cython +/// version compiles down to, to find out how much of the gap is PyO3 and how +/// much is the language. +#[pyfunction] +fn first_consecutive_run_ffi<'py>(seq: &Bound<'py, PyList>) -> PyResult> { + use pyo3::ffi; + let py = seq.py(); + let out = PyList::empty(py); + let raw = seq.as_ptr(); + unsafe { + let n = ffi::PyList_Size(raw); + if n == 0 { + return Ok(out); + } + let mut item = ffi::PyList_GetItem(raw, 0); // borrowed + ffi::PyList_Append(out.as_ptr(), item); + let mut overflow: std::os::raw::c_int = 0; + let mut c_prev = ffi::PyLong_AsLongLongAndOverflow(item, &mut overflow); + if overflow != 0 { + return Ok(out); + } + let mut i: ffi::Py_ssize_t = 1; + while i < n { + item = ffi::PyList_GetItem(raw, i); + let c_cur = ffi::PyLong_AsLongLongAndOverflow(item, &mut overflow); + if overflow != 0 || c_cur <= c_prev || c_cur - 1 != c_prev { + break; + } + ffi::PyList_Append(out.as_ptr(), item); + c_prev = c_cur; + i += 1; + } + } + Ok(out) +} From 63f7f9d5155cde7406e152b7f88e824236091ba0 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 4 Aug 2026 13:07:05 +0000 Subject: [PATCH 4/7] Add CPU and memory tables across Python, PyPy, Cython and Rust The measurements so far compared accelerators inside CPython and only on time. That left out the runtime faust already supports and tests -- PyPy, whose CI leg runs the pure-Python code with USE_CYTHON: 'false' -- and left out memory entirely. extra/tools/bench_accel_matrix.py adds both. Every cell runs in a fresh subprocess, so no implementation's allocations can leak into another's reading, and the PyPy rows are produced by shelling out to pypy3. The PyPy column changes the argument. On HoppingWindow.ranges -- the one case in this whole evaluation where Rust clearly beat Cython, and so the strongest reason to consider it -- PyPy running the pure-Python version is faster than Rust: 162-198 ns against 311 ns, roughly 4x Cython. It has no call boundary to pay, so it JITs ranges together with its caller while both native extensions pay a crossing per call. PyPy does not win the offset scan (0.46-0.90x), where allocation and GC traffic dominate. Memory turns out to support no case at all: on CPython, python, cython and rust land within 0.3 MiB of each other at every size, because all three are dominated by the Python objects the data is made of. PyPy trades a 112 MiB interpreter footprint (6.8x CPython) for a 4.6x more compact list of ints, but peaks far higher (244.8 vs 53.2 MiB at 600k) on GC headroom. Section 3.4 records the tables and the caveats: PyPy here is Python 3.9 against CPython 3.11, and it runs the pure-Python implementations because cpyext makes C extensions slower on PyPy than the Python they replace -- which is why the CI leg disables Cython in the first place. Sections 4 and 6 are updated where the PyPy result changes a verdict. Also corrects the windows figures to ranges rather than point values: PyPy's JIT is the one column with real run-to-run spread. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_019oX4oGCQGgvPJHabjwfaBk --- docs/proposals/rust-acceleration.md | 132 ++++++++++- extra/tools/bench_accel_matrix.py | 354 ++++++++++++++++++++++++++++ 2 files changed, 481 insertions(+), 5 deletions(-) create mode 100644 extra/tools/bench_accel_matrix.py diff --git a/docs/proposals/rust-acceleration.md b/docs/proposals/rust-acceleration.md index 0ec5a7faf..57fd2822d 100644 --- a/docs/proposals/rust-acceleration.md +++ b/docs/proposals/rust-acceleration.md @@ -10,9 +10,9 @@ The prototype works and the build integration is sound, but on the code Faust actually accelerates, Rust lands between 2.3x faster and 15% *slower* than the Cython we already ship — and it is only faster on one method of one class, while being slower on the single busiest loop Faust owns. It would roughly -double the test matrix and add a second native toolchain for that. Prefer keeping Cython, and revisit only if a -hot path appears whose per-element work is *native* rather than -Python-object-bound (see +double the test matrix and add a second native toolchain for that. Prefer +keeping Cython, and revisit only if a hot path appears whose per-element work +is *native* rather than Python-object-bound (see [What would change this answer](#6-what-would-change-this-answer)). > **Update.** The batch-shaped hot path this document was waiting for has @@ -22,6 +22,14 @@ Python-object-bound (see > evaluation: being batch-shaped is not sufficient. See > [§3.3](#33-first_consecutive_run-the-batch-shaped-path-that-arrived), which > also corrects the trigger in §6 that this update fires. +> +> [§3.4](#34-four-way-cpython-pypy-cython-rust--cpu-and-memory) then widens +> the comparison to PyPy and to memory, and finds the strongest single +> argument against this proposal: **on `ranges`, the one case where Rust +> clearly beat Cython, PyPy beats Rust** — 162–198 ns against 311 ns — using +> the pure-Python code already in the tree, on a runtime Faust already tests. +> And memory is indistinguishable between Python, Cython and Rust on CPython, +> so it supports no case either way. Everything below was measured or built, not estimated; see [Reproducing](#7-reproducing). @@ -240,11 +248,106 @@ Two corollaries worth carrying forward: was written: even where abi3 is available, taking it costs ~22% on this kind of code. +### 3.4 Four-way: CPython, PyPy, Cython, Rust — CPU and memory + +Everything above compares accelerators *inside CPython*. That leaves out the +runtime Faust already supports and already tests: PyPy, whose CI leg runs with +`USE_CYTHON: 'false'`, i.e. the pure-Python implementations under its JIT. +It also leaves out memory entirely. + +Measured with `extra/tools/bench_accel_matrix.py`, which runs every cell in a +**fresh subprocess** so no implementation's allocations can leak into another's +reading. CPython 3.11.15, PyPy 7.3.15 (Python 3.9.18), Rust built `--no-abi3` +(its best case, per §3.3). + +#### CPU + +`first_consecutive_run`, the offset commit scan (lower is better): + +| offsets | python | pypy | cython | rust | vs cython | +| ---: | ---: | ---: | ---: | ---: | --- | +| 10 000 | 270.5 µs | 59.4 µs | **54.0 µs** | 63.7 µs | python 0.20x · pypy 0.91x · rust 0.85x | +| 100 000 | 2729.8 µs | 1307.7 µs | **555.5 µs** | 684.0 µs | python 0.20x · pypy 0.42x · rust 0.81x | +| 600 000 | 16843.6 µs | 6158.7 µs | **3416.7 µs** | 3965.8 µs | python 0.20x · pypy 0.55x · rust 0.86x | + +`HoppingWindow.ranges`, per call (lower is better): + +| case | python | pypy | cython | rust | vs cython | +| --- | ---: | ---: | ---: | ---: | --- | +| `ranges(ts)` | 1435–1482 ns | **162–198 ns** | 732–744 ns | 311–317 ns | python 0.50x · **pypy 3.8–4.5x** · rust 2.35x | + +Ranges rather than point values here because PyPy's JIT is the one column with +real run-to-run spread (162–198 ns over four runs; the other three vary by +about 1%). The sign is not in doubt at any point in that band. + +**PyPy wins the window workload outright — around 4x Cython and roughly 1.7x +Rust — and it is the one row in this entire document where something beats +Rust at Rust's best case.** The reason is the same call-boundary argument from +§3.1 read the other way: PyPy has no boundary to pay. It JITs `ranges` and its +caller together into one trace, while both native extensions pay a crossing +per call. +Rust's 1.7–2.3x win over Cython on `ranges` was the single strongest argument +in this document for adopting it; on the interpreter Faust already ships CI +for, that win is not just matched but beaten, for zero build cost. + +PyPy does *not* win the offset scan (0.42–0.91x), and it degrades as the list +grows — the JIT handles the loop well, but the allocation and GC traffic of +building a 600k-element result list does not trace away. + +#### Memory + +Peak RSS of a fresh process, in MiB. "after import" is the interpreter plus +the accelerator module; "input list" is the cost of the `list(range(n))` the +scan runs over; "peak" is the high-water mark for the whole process. + +| offsets | impl | after import | input list | peak | +| ---: | --- | ---: | ---: | ---: | +| 100 000 | python | 16.5 | 3.9 | 22.7 | +| 100 000 | pypy | 112.7 | **1.1** | 135.8 | +| 100 000 | cython | 16.5 | 3.9 | 22.7 | +| 100 000 | rust | 16.7 | 3.9 | 23.0 | +| 600 000 | python | 16.5 | 23.0 | 53.2 | +| 600 000 | pypy | 112.8 | **5.0** | 244.8 | +| 600 000 | cython | 16.5 | 22.9 | **53.2** | +| 600 000 | rust | 16.6 | 23.0 | 53.4 | + +Three things fall out of this, and the first is the one that matters for the +decision: + +* **On CPython the accelerator is invisible.** Python, Cython and Rust land + within 0.3 MiB of each other at every size. Whatever case exists for Rust, + *memory is not part of it* — and neither is it part of the case for Cython. + All three are dominated by the Python objects the data is made of. +* **PyPy trades a fixed cost for a variable one.** Its interpreter footprint + is 112 MiB against CPython's 16.5 MiB (6.8x), but it stores a list of ints + unboxed, so the same 600k offsets cost 5.0 MiB instead of 23.0 MiB (4.6x + less). Its *peak* is nonetheless far worse (244.8 MiB vs 53.2 MiB), which is + GC headroom rather than live data. For a worker holding many partitions' + offset lists the compact representation is the interesting half; for a + memory-constrained deployment the peak is. +* **Artifact size is a wash.** Rust's single `_accel.so` is 498 KiB against + 224 KiB for `faust/utils/_cython/functional` and 541 KiB for + `faust/_cython/windows` — and Faust ships six Cython modules to Rust's one, + so a full replacement would likely shrink the distribution slightly. That is + a real but very small point in Rust's favour, and the only one memory or + size offers. + +#### Caveats + +* PyPy 7.3.15 implements Python 3.9; CPython here is 3.11. That is not a + matched language version, and it is also exactly the version skew the + project actually has, since PyPy trails CPython releases. +* PyPy runs the **pure-Python** implementations. It cannot usefully run the + Cython or Rust modules: `cpyext` makes C extensions slower on PyPy than the + Python they replace, which is why the CI leg sets `USE_CYTHON: 'false'` in + the first place. So "pypy" here is a different *runtime* choice, not a + fourth accelerator — which is the point of including it. + ## 4. Which Faust code could actually go to Rust | Candidate | Shape | Verdict | | --- | --- | --- | -| `faust/_cython/windows.pyx` (109 lines) | Pure float math, no Python objects held | **Portable, mostly not worth it.** The port exists and is correct; §3.1 shows it wins only on `ranges`. | +| `faust/_cython/windows.pyx` (109 lines) | Pure float math, no Python objects held | **Portable, and now clearly not worth it.** The port exists and is correct; §3.1 shows it wins only on `ranges`, and §3.4 shows PyPy running the *pure-Python* version beats that win by ~1.7x. If this shape is worth optimising, the answer is a runtime, not a language. | | `faust/_cython/streams.pyx` (198 lines) | `async def next()`, awaits `chan_slow_get`, `maybe_async`, sensor callbacks | **Poor fit.** The body is an await-driven sequence of calls back into Python; PyO3 needs `pyo3-async-runtimes` to express it and pays the 79 ns boundary on *every* callback it drives. The Cython version wins here precisely because `cdef class` attribute access is cheap, which is the one thing Rust is worse at. | | `faust/transport/_cython/conductor.pyx` (134 lines) | Same — async dispatch, all work is calling Python | **Poor fit**, same reason. | | JSON encode/decode (`faust/utils/json.py`) | Genuinely batch-shaped | **Already solved** by `faust[orjson]`, which is Rust. Writing our own would be strictly worse. | @@ -337,6 +440,12 @@ Concrete triggers, in rough order of likelihood: `streams.pyx` + `conductor.pyx`. If it is small, the entire accelerator question — Rust or Cython — is the wrong thing to optimise. This is the cheapest next step and should precede any language decision. + §3.4 adds a corollary that should be checked at the same time: for the one + shape where a native accelerator looked clearly worthwhile, **PyPy was + faster than both of them**, and it needs no extension at all. Before + reaching for a second toolchain, it is worth knowing what an end-to-end + Faust worker actually costs on PyPy — a question this document cannot + answer from microbenchmarks, and nobody has published either. 3. **Free-threaded CPython becomes a target.** If dropping the `cp31?t-*` skip matters, a `Py_GIL_DISABLED`-safe accelerator is needed and PyO3 is a better starting point than auditing the `.pyx` files. That argues for @@ -363,12 +472,25 @@ pip install -e . # builds the Cython extensions python extra/tools/bench_accel_windows.py # §3.1 python extra/tools/bench_accel_offsets.py # §3.3 +python extra/tools/bench_accel_matrix.py # §3.4 (CPU + memory, incl. PyPy) -# optional: add the rust columns to both +# optional: add the rust columns to all three extra/tools/build_rust_accel.sh # abi3, as a real PR would ship extra/tools/build_rust_accel.sh --no-abi3 # adds the macro-based variants ``` +`bench_accel_matrix.py` shells out to `pypy3` for its PyPy rows and runs every +cell in a fresh subprocess, so it needs the pure-Python dependency chain +importable under PyPy: + +```bash +pypy3 -m pip install mode-streaming yarl aiohttp aiohttp_cors \ + terminaltables croniter mypy_extensions venusian intervaltree six +``` + +Without that (or without `pypy3` at all) the PyPy rows are dropped and the +reason is printed under "Skipped"; the rest of the matrix still runs. + Both benchmarks skip any implementation they cannot import, so they are useful on a pure-Python install too, and `bench_accel_offsets.py` refuses to report timings for implementations that disagree on its correctness cases. diff --git a/extra/tools/bench_accel_matrix.py b/extra/tools/bench_accel_matrix.py new file mode 100644 index 000000000..e24d7ac25 --- /dev/null +++ b/extra/tools/bench_accel_matrix.py @@ -0,0 +1,354 @@ +#!/usr/bin/env python +"""CPU and memory comparison across Python, PyPy, Cython and Rust. + +``bench_accel_windows.py`` and ``bench_accel_offsets.py`` compare +implementations inside one interpreter. This one adds the two axes those +cannot cover: + +* **PyPy**, which needs a separate interpreter (it runs the pure-Python code + under its JIT -- exactly what Faust's CI does, since the PyPy leg sets + ``USE_CYTHON: 'false'``). +* **Memory**, which needs a fresh process per measurement so that peak RSS + means something. + +Every number is produced by a subprocess running exactly one implementation +against one workload, so nothing another implementation allocated can leak +into a reading. + +Usage:: + + pip install -e . # builds the Cython extensions + extra/tools/build_rust_accel.sh --no-abi3 # optional: adds the rust rows + python extra/tools/bench_accel_matrix.py + +Rows for implementations that cannot be imported are dropped with a note, so +this is useful on a pure-Python install and without PyPy installed. +""" + +import argparse +import json +import os +import resource +import shutil +import subprocess +import sys +import timeit +from typing import Any, Callable, Dict, List, Optional, Tuple + +HERE = os.path.dirname(os.path.abspath(__file__)) +REPO = os.path.dirname(os.path.dirname(HERE)) + +#: (label, interpreter, how to get the callable). "pypy" runs the same +#: pure-Python code as "python", on a different interpreter. +IMPLEMENTATIONS = ("python", "pypy", "cython", "rust") + +OFFSET_SIZES = (10_000, 100_000, 600_000) +MIN_TIME = 0.2 +REPEAT = 5 + + +# -------------------------------------------------------------------------- +# workloads, resolved inside the worker process +# -------------------------------------------------------------------------- + + +def _load_offsets(impl: str) -> Callable[[List[int]], Any]: + if impl in ("python", "pypy"): + from faust.utils.functional import _py_first_consecutive_run + + return _py_first_consecutive_run + if impl == "cython": + from faust.utils._cython.functional import first_consecutive_run + + return first_consecutive_run + if impl == "rust": + from faust._rust import _accel + + # The best Rust can do: raw ffi + the CPython macros, no abi3. + return getattr( + _accel, + "first_consecutive_run_macro", + _accel.first_consecutive_run, + ) + raise ValueError(impl) + + +def _load_windows(impl: str) -> Callable[[], Any]: + if impl in ("python", "pypy"): + from faust.windows import _PyHoppingWindow as Window + elif impl == "cython": + from faust._cython.windows import HoppingWindow as Window + elif impl == "rust": + from faust._rust._accel import HoppingWindow as Window + else: + raise ValueError(impl) + return Window(60.0, 10.0, 3600.0) + + +# -------------------------------------------------------------------------- +# worker: one implementation, one workload, one process +# -------------------------------------------------------------------------- + + +def _peak_rss_kb() -> int: + """Peak RSS of this process in KiB (monotonic, so deltas are safe).""" + return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss + + +def _time_ns(fn: Callable[[], Any]) -> float: + # PyPy needs the JIT warmed before the timed run or the first numbers are + # interpreted rather than compiled. + for _ in range(1000): + fn() + number = 1 + while timeit.timeit(fn, number=number) < MIN_TIME: + number *= 4 + best = min(timeit.timeit(fn, number=number) for _ in range(REPEAT)) + return best / number * 1e9 + + +def _worker(impl: str, workload: str, size: int, mode: str) -> Dict[str, Any]: + if workload == "offsets": + run = _load_offsets(impl) + baseline = _peak_rss_kb() + data = list(range(size)) + after_input = _peak_rss_kb() + + def call() -> Any: + return run(data) + + elif workload == "windows": + window = _load_windows(impl) + baseline = _peak_rss_kb() + after_input = baseline + timestamp = 1_700_000_000.123 + + def call() -> Any: + return window.ranges(timestamp) + + else: + raise ValueError(workload) + + out: Dict[str, Any] = {"impl": impl, "workload": workload, "size": size} + if mode == "cpu": + out["ns"] = _time_ns(call) + else: + # Keep the results alive across the reading, so the peak includes + # what the call produced rather than just what it was given. + held = [call() for _ in range(3)] + out["baseline_kb"] = baseline + out["input_kb"] = after_input - baseline + out["peak_kb"] = _peak_rss_kb() + out["result_kb"] = out["peak_kb"] - after_input + out["held"] = len(held) + return out + + +# -------------------------------------------------------------------------- +# driver +# -------------------------------------------------------------------------- + + +def _interpreter_for(impl: str) -> Optional[str]: + if impl == "pypy": + return shutil.which("pypy3") + return sys.executable + + +def _run_one( + impl: str, workload: str, size: int, mode: str +) -> Tuple[Optional[Dict[str, Any]], str]: + interpreter = _interpreter_for(impl) + if interpreter is None: + return None, "pypy3 not on PATH" + # PyPy has no editable install of faust, so put the repo on its path. + env = dict(os.environ) + env["PYTHONPATH"] = os.pathsep.join(filter(None, [REPO, env.get("PYTHONPATH", "")])) + proc = subprocess.run( + [ + interpreter, + os.path.abspath(__file__), + "--worker", + "--impl", + impl, + "--workload", + workload, + "--size", + str(size), + "--mode", + mode, + ], + cwd=REPO, + env=env, + capture_output=True, + text=True, + ) + for line in proc.stdout.splitlines(): + if line.startswith("{"): + return json.loads(line), "" + detail = (proc.stderr or proc.stdout).strip().splitlines() + return None, detail[-1][:80] if detail else f"exit {proc.returncode}" + + +def _versions() -> Dict[str, str]: + versions = {"python": f"CPython {sys.version.split()[0]}"} + versions["cython"] = versions["python"] + versions["rust"] = versions["python"] + pypy = shutil.which("pypy3") + if pypy: + out = subprocess.run( + [ + pypy, + "-c", + "import sys; print(sys.version.split()[0], sys.pypy_version_info[:3])", + ], + capture_output=True, + text=True, + ).stdout.split() + if out: + versions["pypy"] = f"PyPy (Python {out[0]})" + return versions + + +def _artifact_sizes() -> List[Tuple[str, str, float]]: + """On-disk size of each accelerator, which is a real deployment cost.""" + candidates = [ + ("cython", "faust/utils/_cython/functional"), + ("cython", "faust/_cython/windows"), + ("rust", "faust/_rust/_accel"), + ] + found = [] + for label, stem in candidates: + for suffix in (".so", ".abi3.so", ".cpython-311-x86_64-linux-gnu.so"): + path = os.path.join(REPO, stem + suffix) + if os.path.exists(path): + found.append((label, stem.split("/")[-1], os.path.getsize(path) / 1024)) + break + return found + + +def _ratios(results: Dict[str, float]) -> str: + """Speed relative to Cython, which is the thing that ships today.""" + if "cython" not in results: + return "-" + base = results["cython"] + return " ".join( + f"{impl}={base / results[impl]:.2f}x" + for impl in IMPLEMENTATIONS + if impl in results and impl != "cython" + ) + + +def _table(title: str, header: List[str], rows: List[List[str]]) -> None: + widths = [ + max(len(str(header[i])), max((len(str(r[i])) for r in rows), default=0)) + for i in range(len(header)) + ] + print(f"\n{title}") + print(" ".join(h.ljust(widths[i]) for i, h in enumerate(header))) + print("-" * (sum(widths) + 2 * (len(widths) - 1))) + for row in rows: + print(" ".join(str(c).ljust(widths[i]) for i, c in enumerate(row))) + + +def main() -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--worker", action="store_true", help=argparse.SUPPRESS) + parser.add_argument("--impl") + parser.add_argument("--workload") + parser.add_argument("--size", type=int, default=0) + parser.add_argument("--mode", default="cpu") + args = parser.parse_args() + + if args.worker: + print(json.dumps(_worker(args.impl, args.workload, args.size, args.mode))) + return 0 + + versions = _versions() + print("Interpreters and builds:") + for impl in IMPLEMENTATIONS: + print(f" {impl:>7}: {versions.get(impl, 'not available')}") + + skipped: Dict[str, str] = {} + + # ---- CPU: offsets ---- + rows = [] + for size in OFFSET_SIZES: + row = [f"{size:,}"] + results = {} + for impl in IMPLEMENTATIONS: + result, why = _run_one(impl, "offsets", size, "cpu") + if result is None: + skipped.setdefault(impl, why) + row.append("-") + else: + results[impl] = result["ns"] + row.append(f"{result['ns'] / 1000:,.1f} us") + row.append(_ratios(results)) + rows.append(row) + _table( + "CPU -- first_consecutive_run (offset commit scan), lower is better", + ["offsets", "python", "pypy", "cython", "rust", "vs cython"], + rows, + ) + + # ---- CPU: windows ---- + rows = [] + row = ["HoppingWindow.ranges"] + results = {} + for impl in IMPLEMENTATIONS: + result, why = _run_one(impl, "windows", 0, "cpu") + if result is None: + skipped.setdefault(impl, why) + row.append("-") + else: + results[impl] = result["ns"] + row.append(f"{result['ns']:,.0f} ns") + row.append(_ratios(results)) + rows.append(row) + _table( + "CPU -- HoppingWindow.ranges (per call), lower is better", + ["case", "python", "pypy", "cython", "rust", "vs cython"], + rows, + ) + + # ---- Memory ---- + rows = [] + for size in (100_000, 600_000): + for impl in IMPLEMENTATIONS: + result, why = _run_one(impl, "offsets", size, "mem") + if result is None: + skipped.setdefault(impl, why) + continue + rows.append( + [ + f"{size:,}", + impl, + f"{result['baseline_kb'] / 1024:,.1f}", + f"{result['input_kb'] / 1024:,.1f}", + f"{result['peak_kb'] / 1024:,.1f}", + ] + ) + _table( + "Memory -- fresh process per row, peak RSS in MiB", + ["offsets", "impl", "after import", "input list", "peak"], + rows, + ) + + artifacts = _artifact_sizes() + if artifacts: + _table( + "Artifact size on disk (KiB)", + ["kind", "module", "size"], + [[k, m, f"{s:,.0f}"] for k, m, s in artifacts], + ) + + if skipped: + print("\nSkipped:") + for impl, why in skipped.items(): + print(f" {impl}: {why}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) From 2f04895ed5138087eae1f54a7d813bb7b37816d9 Mon Sep 17 00:00:00 2001 From: William Barnhart Date: Tue, 21 Jul 2026 15:25:26 +0000 Subject: [PATCH 5/7] test(assignor): disable Hypothesis deadline to fix flaky PyPy CI leg 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 Claude-Session: https://claude.ai/code/session_01HHPL4VFWQRQPpjR1gXSKyL --- tests/meticulous/assignor/test_copartitioned_assignor.py | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) diff --git a/tests/meticulous/assignor/test_copartitioned_assignor.py b/tests/meticulous/assignor/test_copartitioned_assignor.py index e0fb1f66b..d78ceea8c 100644 --- a/tests/meticulous/assignor/test_copartitioned_assignor.py +++ b/tests/meticulous/assignor/test_copartitioned_assignor.py @@ -8,7 +8,13 @@ from faust.assignor.client_assignment import CopartitionedAssignment from faust.assignor.copartitioned_assignor import CopartitionedAssignor -TEST_DEADLINE = 4000 +# These Hypothesis property tests assert assignment *correctness*, not +# performance. PyPy's JIT warmup makes a single example's timing vary by +# well over a second, so a fixed deadline is tripped intermittently +# (DeadlineExceeded -> FlakyFailure) even though the assignment is valid -- +# flaking the PyPy CI leg. Disable the deadline; the job's own timeout still +# guards against a real hang. +TEST_DEADLINE = None _topics = {"foo", "bar", "baz"} From 941a2199097bfe9866cb0ceb678f09080f1a1d08 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 4 Aug 2026 13:43:57 +0000 Subject: [PATCH 6/7] Tighten static typing in the Cython accelerators 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 Claude-Session: https://claude.ai/code/session_019oX4oGCQGgvPJHabjwfaBk --- faust/sensors/_cython/base.pyx | 4 +-- faust/transport/_cython/scheduler.pyx | 39 +++++++++++++++++++-------- 2 files changed, 30 insertions(+), 13 deletions(-) diff --git a/faust/sensors/_cython/base.pyx b/faust/sensors/_cython/base.pyx index 72a960631..6b16282fb 100644 --- a/faust/sensors/_cython/base.pyx +++ b/faust/sensors/_cython/base.pyx @@ -19,13 +19,13 @@ cdef class SensorDelegateBase: self.app = app self._sensors = set() - def add(self, sensor): + def add(self, object sensor): """Add sensor.""" # connect beacons sensor.beacon = self.app.beacon.new(sensor) self._sensors.add(sensor) - def remove(self, sensor): + def remove(self, object sensor): """Remove sensor.""" self._sensors.remove(sensor) diff --git a/faust/transport/_cython/scheduler.pyx b/faust/transport/_cython/scheduler.pyx index 5d153c02d..e62e51ab2 100644 --- a/faust/transport/_cython/scheduler.pyx +++ b/faust/transport/_cython/scheduler.pyx @@ -81,21 +81,31 @@ cdef class _TopicCursor: continue return (PyList_GET_ITEM(self.tps, i), item) - cdef _start_pass(self): + cdef int _start_pass(self) except -1: + # The Python calls below (dict.pop, .items()) can raise, and this + # says so explicitly rather than relying on a compiler default: + # Cython 3 propagates from a bare `cdef` by default, but Cython 0.x + # silently swallowed unless an except clause was given, and + # `legacy_implicit_noexcept` restores that. `except -1` propagates + # under every version. cdef: object tp object it + Py_ssize_t live if self.to_remove: for tp in self.to_remove: self.buffers.pop(tp, None) self.to_remove.clear() - elif PyList_GET_SIZE(self.tps) == len(self.buffers): - # Nothing drained and nothing added since the last pass, so the - # snapshot is still accurate. TopicBuffer.add() asserts the - # partition is new, so the size can only change on a real change. - self.pi = 0 - return + else: + live = len(self.buffers) + if PyList_GET_SIZE(self.tps) == live: + # Nothing drained and nothing added since the last pass, so + # the snapshot is still accurate. TopicBuffer.add() asserts + # the partition is new, so the size can only change on a + # real change. + self.pi = 0 + return 0 self.tps = [] self.iters = [] for tp, it in self.buffers.items(): @@ -103,6 +113,7 @@ cdef class _TopicCursor: self.iters.append(it) self.n = PyList_GET_SIZE(self.tps) self.pi = 0 + return 0 cdef class RoundRobinRecordIterator: @@ -164,20 +175,25 @@ cdef class RoundRobinRecordIterator: continue return item - cdef _start_pass(self): + cdef int _start_pass(self) except -1: + # See the note on _TopicCursor._start_pass for why the except clause + # is spelled out. cdef: object topic object buffer _TopicCursor cursor + Py_ssize_t live if self.to_remove: for topic in self.to_remove: self.index.pop(topic, None) self.cursors.pop(topic, None) self.to_remove.clear() - elif PyList_GET_SIZE(self.topics) == len(self.index): - self.ti = 0 - return + else: + live = len(self.index) + if PyList_GET_SIZE(self.topics) == live: + self.ti = 0 + return 0 self.topics = [] self.topic_cursors = [] for topic, buffer in self.index.items(): @@ -189,6 +205,7 @@ cdef class RoundRobinRecordIterator: self.topic_cursors.append(cursor) self.n = PyList_GET_SIZE(self.topics) self.ti = 0 + return 0 cpdef object records_iterator(object index): From 340793fa4bf43bd35a60d7fe27bfed81160b75e4 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 4 Aug 2026 14:09:44 +0000 Subject: [PATCH 7/7] Fix records being dropped when a topic buffer is replaced mid-iteration 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 Claude-Session: https://claude.ai/code/session_019oX4oGCQGgvPJHabjwfaBk --- faust/transport/_cython/scheduler.pyx | 76 +++++++++++++++++++++------ tests/unit/transport/test_utils.py | 38 ++++++++++++++ 2 files changed, 98 insertions(+), 16 deletions(-) diff --git a/faust/transport/_cython/scheduler.pyx b/faust/transport/_cython/scheduler.pyx index e62e51ab2..43475b154 100644 --- a/faust/transport/_cython/scheduler.pyx +++ b/faust/transport/_cython/scheduler.pyx @@ -91,21 +91,16 @@ cdef class _TopicCursor: cdef: object tp object it - Py_ssize_t live if self.to_remove: for tp in self.to_remove: self.buffers.pop(tp, None) self.to_remove.clear() - else: - live = len(self.buffers) - if PyList_GET_SIZE(self.tps) == live: - # Nothing drained and nothing added since the last pass, so - # the snapshot is still accurate. TopicBuffer.add() asserts - # the partition is new, so the size can only change on a - # real change. - self.pi = 0 - return 0 + elif self._snapshot_is_current(): + # Unchanged since the last pass, so the existing snapshot still + # describes _buffers exactly and can be reused. + self.pi = 0 + return 0 self.tps = [] self.iters = [] for tp, it in self.buffers.items(): @@ -115,6 +110,33 @@ cdef class _TopicCursor: self.pi = 0 return 0 + cdef bint _snapshot_is_current(self) except -1: + """Is the cached snapshot still identical to the live buffer map? + + Comparing lengths alone is not enough: an iterator swapped in under + an existing TP keeps the length the same but leaves a stale cursor, + whose records would never be delivered. So the entries are compared + by identity. That walks the mapping, but allocates nothing, which is + what makes reusing the snapshot worth doing at all. + """ + cdef: + Py_ssize_t i = 0 + Py_ssize_t n = PyList_GET_SIZE(self.tps) + object tp + object it + + if n != len(self.buffers): + return False + for tp, it in self.buffers.items(): + if i >= n: + return False + if PyList_GET_ITEM(self.tps, i) is not tp: + return False + if PyList_GET_ITEM(self.iters, i) is not it: + return False + i += 1 + return i == n + cdef class RoundRobinRecordIterator: """Iterate a topic index map in round-robin order. @@ -182,18 +204,15 @@ cdef class RoundRobinRecordIterator: object topic object buffer _TopicCursor cursor - Py_ssize_t live if self.to_remove: for topic in self.to_remove: self.index.pop(topic, None) self.cursors.pop(topic, None) self.to_remove.clear() - else: - live = len(self.index) - if PyList_GET_SIZE(self.topics) == live: - self.ti = 0 - return 0 + elif self._snapshot_is_current(): + self.ti = 0 + return 0 self.topics = [] self.topic_cursors = [] for topic, buffer in self.index.items(): @@ -207,6 +226,31 @@ cdef class RoundRobinRecordIterator: self.ti = 0 return 0 + cdef bint _snapshot_is_current(self) except -1: + """Is the cached snapshot still identical to the live index map? + + As with _TopicCursor, a length check alone would miss a TopicBuffer + replaced under an existing topic name, so each cursor is compared + against the buffer it was built from. + """ + cdef: + Py_ssize_t i = 0 + Py_ssize_t n = PyList_GET_SIZE(self.topics) + object topic + object buffer + + if n != len(self.index): + return False + for topic, buffer in self.index.items(): + if i >= n: + return False + if PyList_GET_ITEM(self.topics, i) is not topic: + return False + if (<_TopicCursor>PyList_GET_ITEM(self.topic_cursors, i)).source is not buffer: + return False + i += 1 + return i == n + cpdef object records_iterator(object index): """Iterate over a topic index map in round-robin order.""" diff --git a/tests/unit/transport/test_utils.py b/tests/unit/transport/test_utils.py index 66bda8d2d..1f92597c6 100644 --- a/tests/unit/transport/test_utils.py +++ b/tests/unit/transport/test_utils.py @@ -206,3 +206,41 @@ def test_drains_the_buffer_map_too(self, impl): list(impl(index)) assert not index assert not buffer._buffers + + @pytest.mark.parametrize("impl", RECORDS_ITERATOR_IMPLS) + def test_topic_buffer_replaced_mid_iteration(self, impl): + # A TopicBuffer swapped in under an existing topic name keeps the + # index the same size, so a length-only staleness check would miss + # it and keep draining the old buffer -- silently dropping every + # record in the replacement. + first = TopicBuffer() + first.add(TP1, [1, 2]) + index = {"foo": first} + + it = impl(index) + assert next(it) == (TP1, 1) + + replacement = TopicBuffer() + replacement.add(TP1, [10, 11]) + index["foo"] = replacement + + assert list(it) == [(TP1, 10), (TP1, 11)] + + @pytest.mark.parametrize("impl", RECORDS_ITERATOR_IMPLS) + def test_partition_buffer_replaced_mid_iteration(self, impl): + # Same hazard one level down: an iterator swapped in under an + # existing TP leaves the partition cursor stale. + buffer = TopicBuffer() + buffer.add(TP1, [1, 2, 3]) + + it = impl({"foo": buffer}) + assert next(it) == (TP1, 1) + + buffer._buffers[TP1] = iter([10, 11]) + assert list(it) == [(TP1, 10), (TP1, 11)] + + # Note: swapping a topic for one under a *different* name mid-iteration + # is deliberately not covered. The pure-Python generator iterates the + # live dict, so CPython raises "dictionary keys changed during + # iteration" -- that is undefined behaviour in Python itself, not a + # guarantee either implementation should be pinned to.