diff --git a/.gitignore b/.gitignore index e6974ba9a..2f1d3b601 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 @@ -50,3 +52,4 @@ faust/transport/_cython/*.c # coverage .coverage coverage.xml +faust/_rust/target/ diff --git a/docs/proposals/rust-acceleration.md b/docs/proposals/rust-acceleration.md new file mode 100644 index 000000000..57fd2822d --- /dev/null +++ b/docs/proposals/rust-acceleration.md @@ -0,0 +1,507 @@ +# 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 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. +> +> [§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). + +--- + +## 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.) + +**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 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 **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 + +| 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. + +### 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. + +### 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, 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. | +| 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. | +| `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 +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 + +**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**~~ — **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 + 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 + *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 (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 # §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 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. + +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_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()) 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/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() 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/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) +} 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..6b16282fb --- /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, object sensor): + """Add sensor.""" + # connect beacons + sensor.beacon = self.app.beacon.new(sensor) + self._sensors.add(sensor) + + def remove(self, object 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..43475b154 --- /dev/null +++ b/faust/transport/_cython/scheduler.pyx @@ -0,0 +1,257 @@ +# 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 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 + + if self.to_remove: + for tp in self.to_remove: + self.buffers.pop(tp, None) + self.to_remove.clear() + 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(): + self.tps.append(tp) + self.iters.append(it) + self.n = PyList_GET_SIZE(self.tps) + 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. + + 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 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 + + 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 self._snapshot_is_current(): + self.ti = 0 + return 0 + 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 + 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.""" + 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/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"} 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..1f92597c6 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,156 @@ 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 + + @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. 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