Implement native protocol parsing

.github/workflows/ci.yaml

@@ -24,6 +24,34 @@ jobs:
     with:
       python-version: "3.13"
 
+  rust-lint:
+    name: Rust static analysis
+    runs-on: ubuntu-latest
+    defaults:
+      run:
+        working-directory: rust
+    steps:
+      - uses: actions/checkout@v6
+
+      - name: Install Rust toolchain
+        uses: dtolnay/rust-toolchain@stable
+        with:
+          components: rustfmt, clippy
+
+      - name: Cache Rust build artefacts
+        uses: Swatinem/rust-cache@v2
+        with:
+          workspaces: rust
+
+      - name: Check formatting
+        run: cargo fmt --check
+
+      - name: Clippy
+        run: cargo clippy -- -D warnings
+
+      - name: Test
+        run: cargo test
+
   check-build:
     name: Check packaging metadata
     uses: less-action/reusables/.github/workflows/python-test-build.yaml@f26f5812fa62d31b9a4e0af9f75f19f7f028425c
@@ -37,6 +65,14 @@ jobs:
     steps:
       - uses: actions/checkout@v6
 
+      - name: Install Rust toolchain
+        uses: dtolnay/rust-toolchain@stable
+
+      - name: Cache Rust build artefacts
+        uses: Swatinem/rust-cache@v2
+        with:
+          workspaces: rust
+
       - name: Set up Python
         uses: actions/setup-python@v6
         with:
@@ -72,6 +108,14 @@ jobs:
     steps:
       - uses: actions/checkout@v6
 
+      - name: Install Rust toolchain
+        uses: dtolnay/rust-toolchain@stable
+
+      - name: Cache Rust build artefacts
+        uses: Swatinem/rust-cache@v2
+        with:
+          workspaces: rust
+
       - name: Set up Python
         uses: actions/setup-python@v6
         with:
@@ -94,6 +138,14 @@ jobs:
     steps:
       - uses: actions/checkout@v6
 
+      - name: Install Rust toolchain
+        uses: dtolnay/rust-toolchain@stable
+
+      - name: Cache Rust build artefacts
+        uses: Swatinem/rust-cache@v2
+        with:
+          workspaces: rust
+
       - name: Set up Python
         uses: actions/setup-python@v6
         with:
@@ -116,6 +168,14 @@ jobs:
     steps:
       - uses: actions/checkout@v6
 
+      - name: Install Rust toolchain
+        uses: dtolnay/rust-toolchain@stable
+
+      - name: Cache Rust build artefacts
+        uses: Swatinem/rust-cache@v2
+        with:
+          workspaces: rust
+
       - name: Start Kafka backend
         run: docker compose up -d
 
@@ -196,6 +256,6 @@ jobs:
       - name: Install dependencies
         run: pip install --require-hashes -r docs/requirements.txt
       - name: Install pushed version
-        run: pip install --no-dependencies .
+        run: pip install .
       - name: Build docs
         run: sphinx-build -Wb html docs docs/_build

.gitignore

@@ -142,3 +142,6 @@ dmypy.json
 /schema/
 error-codes.txt
 src/kio/_version.py
+
+# Maturin
+src/kio/lib_kio_native.dylib.dSYM/

.pre-commit-config.yaml

@@ -17,6 +17,15 @@ repos:
         args:
           - -disable-indent-size
 
+  - repo: local
+    hooks:
+      - id: cargo-fmt
+        name: cargo fmt
+        entry: cargo fmt --manifest-path rust/Cargo.toml --
+        language: system
+        types: [rust]
+        pass_filenames: false
+
   - repo: https://github.com/charliermarsh/ruff-pre-commit
     rev: "v0.15.6"
     hooks:
@@ -43,7 +52,7 @@ repos:
           - typing-extensions==4.10.0
           - pytest==7.4.3
           - pydantic==1.10.13
-          - hypothesis==6.99.12
+          - hypothesis==6.151.9
           - types-setuptools==78.1.0.20250329
           - types-requests==2.31.0.20240311
 
@@ -72,6 +81,5 @@ repos:
       - id: check-manifest
         args: ["--no-build-isolation"]
         additional_dependencies:
-          - setuptools==80.9.0
+          - maturin==1.12.5
           - setuptools-scm==9.2.2
-          - wheel==0.45.1

MANIFEST.in

@@ -9,6 +9,7 @@ exclude *.yaml
 exclude *.yml
 exclude Makefile
 exclude setup.cfg
+exclude build_kio.py
 recursive-include src py.typed
 recursive-include src *.py
 include LICENSE

Makefile

@@ -1,3 +1,6 @@
+# Enable globstar to allow ** to match recursive directories.
+SHELL := /usr/bin/env bash -O globstar
+
 .PHONY: fetch-schema-src
 fetch-schema-src:
 	python3 -m codegen.fetch_schema
@@ -11,3 +14,14 @@ generate-schema:
 
 .PHONY: build-schema
 build-schema: fetch-schema-src generate-schema
+
+.PHONY: clean
+clean:
+	rm -rf {**/,}*.egg-info {**/,}/__pycache__ build dist src/{**/,}*.so .coverage rust/target src/kio/_version.py src/kio/lib_kio_native.dylib.dSYM
+
+.PHONY: nuke
+nuke: clean
+	rm -rf .hypothesis .mypy_cache .import_linter_cache .pytest_cache .ruff_cache
+
+build:
+	maturin develop

benchmarks/parsing.py

@@ -0,0 +1,134 @@
+"""
+Large flexible `MetadataResponse` parse micro-benchmark.
+
+Reproduce and compare branches (after rebuilding the extension, e.g. ``maturin develop --release``)::
+
+    PYTHONPATH=src python benchmarks/parsing.py --fast -o /tmp/kio-parse-main.json   # on main
+    PYTHONPATH=src python benchmarks/parsing.py --fast -o /tmp/kio-parse-branch.json # on your branch
+    python -m pyperf compare_to /tmp/kio-parse-main.json /tmp/kio-parse-branch.json
+
+Use ``--rigorous`` instead of ``--fast`` when you need stable numbers; run outside a sandbox
+if ``pyperf`` fails collecting load average metadata.
+"""
+
+from __future__ import annotations
+
+import datetime
+import io
+import uuid
+
+from copy import replace
+
+from kio.schema.errors import ErrorCode
+from kio.schema.metadata.v12 import MetadataResponse
+from kio.schema.metadata.v12.response import MetadataResponseBroker
+from kio.schema.metadata.v12.response import MetadataResponsePartition
+from kio.schema.metadata.v12.response import MetadataResponseTopic
+from kio.schema.types import BrokerId
+from kio.schema.types import TopicName
+from kio.serial import entity_reader
+from kio.serial import entity_writer
+from kio.static.primitive import i32
+from kio.static.primitive import i32Timedelta
+
+write_metadata_response = entity_writer(MetadataResponse)
+read_metadata_response = entity_reader(MetadataResponse)
+
+instance = MetadataResponse(
+    throttle_time=i32Timedelta.parse(datetime.timedelta(milliseconds=123)),
+    brokers=tuple(
+        MetadataResponseBroker(
+            node_id=BrokerId(n),
+            host="foo.bar",
+            port=i32(1234),
+            rack=None,
+        )
+        for n in range(20)
+    ),
+    cluster_id="556",
+    controller_id=BrokerId(3),
+    topics=tuple(
+        MetadataResponseTopic(
+            error_code=ErrorCode.kafka_storage_error,
+            name=TopicName(f"topic {n}"),
+            topic_id=uuid.uuid4(),
+            is_internal=False,
+            partitions=tuple(
+                MetadataResponsePartition(
+                    error_code=ErrorCode.delegation_token_expired,
+                    partition_index=i32(p),
+                    leader_id=BrokerId(2345),
+                    leader_epoch=i32(6445678),
+                    replica_nodes=(BrokerId(12345), BrokerId(7651)),
+                    isr_nodes=(),
+                    offline_replicas=(),
+                )
+                for p in range(12)
+            ),
+            topic_authorized_operations=i32(765443),
+        )
+        for n in range(1_000)
+    ),
+)
+
+with io.BytesIO() as buffer:
+    write_metadata_response(buffer, replace(instance, controller_id=BrokerId(1001)))
+    write_metadata_response(buffer, replace(instance, controller_id=BrokerId(1002)))
+    write_metadata_response(buffer, replace(instance, controller_id=BrokerId(1003)))
+    write_metadata_response(buffer, replace(instance, controller_id=BrokerId(1004)))
+    write_metadata_response(buffer, replace(instance, controller_id=BrokerId(1005)))
+    write_metadata_response(buffer, replace(instance, controller_id=BrokerId(1006)))
+    write_metadata_response(buffer, replace(instance, controller_id=BrokerId(1007)))
+    write_metadata_response(buffer, replace(instance, controller_id=BrokerId(1008)))
+    write_metadata_response(buffer, replace(instance, controller_id=BrokerId(1009)))
+    write_metadata_response(buffer, replace(instance, controller_id=BrokerId(1010)))
+    buffer = buffer.getvalue()
+
+
+def perform_parsing(
+    loops: int,
+) -> float:
+    loop_range = range(loops)
+    accumulated = 0.0
+
+    for _ in loop_range:
+        offset = 0
+        t0 = pyperf.perf_counter()
+        _, size = read_metadata_response(buffer, offset)
+        offset += size
+        _, size = read_metadata_response(buffer, offset)
+        offset += size
+        _, size = read_metadata_response(buffer, offset)
+        offset += size
+        _, size = read_metadata_response(buffer, offset)
+        offset += size
+        _, size = read_metadata_response(buffer, offset)
+        offset += size
+        _, size = read_metadata_response(buffer, offset)
+        offset += size
+        _, size = read_metadata_response(buffer, offset)
+        offset += size
+        _, size = read_metadata_response(buffer, offset)
+        offset += size
+        _, size = read_metadata_response(buffer, offset)
+        offset += size
+        _, size = read_metadata_response(buffer, offset)
+        offset += size
+        accumulated += pyperf.perf_counter() - t0
+        assert offset == len(buffer), "buffer not exhausted after read"
+
+    return accumulated
+
+
+if __name__ == "__main__":
+    # from scalene import scalene_profiler
+    # scalene_profiler.start()
+    # perform_parsing(1000)
+    # scalene_profiler.stop()
+
+    # perform_parsing(1000)
+
+    import pyperf
+
+    runner = pyperf.Runner()
+    runner.bench_time_func("roundtrip", perform_parsing, inner_loops=10000)