Perf/profile guided optimizations

.github/workflows/hotpath.yml

@@ -0,0 +1,58 @@
+name: Hotpath profiling coverage
+
+on:
+  workflow_dispatch:
+  workflow_call:
+
+permissions:
+  contents: read
+
+jobs:
+  profile:
+    runs-on: ubuntu-latest
+    timeout-minutes: 20
+    strategy:
+      fail-fast: false
+      matrix:
+        feature: [hotpath-profile, hotpath-alloc-profile]
+    steps:
+      - uses: actions/checkout@v6
+        with:
+          persist-credentials: false
+      - uses: actions/setup-python@v6
+        with:
+          python-version: '3.14'
+      - uses: dtolnay/rust-toolchain@master
+        with:
+          toolchain: nightly-2026-09-25
+      - uses: Swatinem/rust-cache@v2
+      - uses: astral-sh/setup-uv@v6
+      - name: Install build and test dependencies
+        run: uv sync --locked --no-install-project
+      - name: Audit hooks and test the inventory parser
+        run: |
+          cargo test --locked --manifest-path tools/hotpath-coverage/Cargo.toml
+          cargo run --locked --manifest-path tools/hotpath-coverage/Cargo.toml
+      - name: Build the instrumented extension
+        run: uv run --no-sync maturin develop --release --features ${{ matrix.feature }}
+      - name: Generate TLS fixtures
+        run: uv run --no-sync python scripts/generate_test_tls_certs.py
+      - name: Test profile lifecycle and tooling
+        run: |
+          uv run --no-sync pytest -q tests/test_hotpath_profile.py
+          uv run --no-sync pytest -q -m tooling tests/tooling/test_profile_with_hotpath.py
+      - name: Collect every workload in fresh processes
+        run: >-
+          uv run --no-sync python scripts/profile_with_hotpath.py all
+          --output target/hotpath-ci
+          --callbacks 1000 --tasks 200 --task-batch-size 50
+          --tcp-roundtrips 30 --timers 200 --timer-batch-size 50
+          --connections 5 --probe-iterations 3 --concurrency 2
+          --requests-per-connection 3 --bulk-bytes 131072
+          --idle-connections 4 --idle-cycles 2
+      - uses: actions/upload-artifact@v4
+        if: always()
+        with:
+          name: ${{ matrix.feature }}-${{ github.run_attempt }}
+          path: target/hotpath-ci/
+          retention-days: 14

.github/workflows/tests.yml

@@ -64,6 +64,9 @@ jobs:
   vibeio-native:
     uses: ./.github/workflows/vibeio-native.yml
 
+  hotpath-profile:
+    uses: ./.github/workflows/hotpath.yml
+
   clippy:
     name: clippy (risk-focused)
     runs-on: ubuntu-latest

Cargo.toml

@@ -61,6 +61,8 @@ windows-sys = { version = "0.61", features = ["Wdk_Foundation", "Wdk_Storage_Fil
 [features]
 default = []
 hotpath-profile = ["dep:hotpath", "hotpath/hotpath"]
+# Separate diagnostic build: Rust allocations, not CPython's allocator.
+hotpath-alloc-profile = ["hotpath-profile", "hotpath/hotpath-alloc"]
 # Opt-in: faster scheduler turns, but measured TCP workloads can regress.
 scheduler-batch-cache = []
 blocking-default = ["dep:rusty_pool"]

benches/profile_workloads.py

@@ -0,0 +1,190 @@
+"""Small correctness-checked probes for runtime paths outside TCP benchmarks."""
+
+from __future__ import annotations
+
+import asyncio
+import contextvars
+import os
+import signal
+import socket
+import sys
+import tempfile
+import threading
+from pathlib import Path
+
+
+async def threadsafe(iterations: int) -> None:
+    loop = asyncio.get_running_loop()
+    done = loop.create_future()
+    seen = []
+    value = contextvars.ContextVar("profile_value", default=-1)
+
+    def receive(i):
+        seen.append((i, value.get()))
+        if len(seen) == iterations:
+            done.set_result(None)
+
+    def producer():
+        for i in range(iterations):
+            value.set(i)
+            loop.call_soon_threadsafe(receive, i)
+
+    worker = threading.Thread(target=producer, name="profile-producer")
+    worker.start()
+    try:
+        await done
+    finally:
+        worker.join()
+    assert seen == [(i, i) for i in range(iterations)]
+
+
+async def executor_dns(iterations: int) -> None:
+    loop = asyncio.get_running_loop()
+    for i in range(iterations):
+        assert await loop.run_in_executor(None, sum, range(i + 1)) == i * (i + 1) // 2
+        addresses = await loop.getaddrinfo("localhost", 80, type=socket.SOCK_STREAM)
+        assert addresses
+        host, port = await loop.getnameinfo(
+            ("127.0.0.1", 80), socket.NI_NUMERICHOST | socket.NI_NUMERICSERV
+        )
+        assert (host, port) == ("127.0.0.1", "80")
+
+
+async def udp(iterations: int) -> None:
+    loop = asyncio.get_running_loop()
+    with (
+        socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as left,
+        socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as right,
+    ):
+        left.bind(("127.0.0.1", 0))
+        right.bind(("127.0.0.1", 0))
+        left.setblocking(False)
+        right.setblocking(False)
+        for i in range(iterations):
+            payload = i.to_bytes(8, "little")
+            await loop.sock_sendto(left, payload, right.getsockname())
+            data, addr = await loop.sock_recvfrom(right, 64)
+            assert data == payload
+            await loop.sock_sendto(right, data, addr)
+            buffer = bytearray(64)
+            size, _ = await loop.sock_recvfrom_into(left, buffer)
+            assert buffer[:size] == payload
+
+
+async def socket_ops(iterations: int) -> None:
+    loop = asyncio.get_running_loop()
+    with socket.socket() as listener, socket.socket() as client:
+        listener.bind(("127.0.0.1", 0))
+        listener.listen()
+        listener.setblocking(False)
+        client.setblocking(False)
+        connecting = asyncio.ensure_future(
+            loop.sock_connect(client, listener.getsockname())
+        )
+        server, _ = await loop.sock_accept(listener)
+        await connecting
+        with server:
+            server.setblocking(False)
+            for _ in range(iterations):
+                await loop.sock_sendall(client, b"socket probe")
+                data = await loop.sock_recv(server, 64)
+                assert data == b"socket probe"
+                await loop.sock_sendall(server, data)
+                buffer = bytearray(64)
+                size = await loop.sock_recv_into(client, buffer)
+                assert buffer[:size] == data
+
+
+async def subprocess_pipes(iterations: int) -> None:
+    for _ in range(iterations):
+        process = await asyncio.create_subprocess_exec(
+            sys.executable,
+            "-c",
+            "import sys; sys.stdout.buffer.write(sys.stdin.buffer.read()); sys.stderr.write('err')",
+            stdin=asyncio.subprocess.PIPE,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        try:
+            out, err = await process.communicate(b"pipe probe" * 1024)
+            assert out == b"pipe probe" * 1024 and err == b"err"
+            assert process.returncode == 0
+        finally:
+            if process.returncode is None:
+                process.kill()
+                await process.wait()
+
+
+async def unix_streams(iterations: int) -> None:
+    async def echo(reader, writer):
+        try:
+            while data := await reader.read(4096):
+                writer.write(data)
+                await writer.drain()
+        finally:
+            writer.close()
+            await writer.wait_closed()
+
+    with tempfile.TemporaryDirectory(prefix="rsloop-profile-") as directory:
+        path = str(Path(directory) / "sock")
+        server = await asyncio.start_unix_server(echo, path)
+        try:
+            reader, writer = await asyncio.open_unix_connection(path)
+            try:
+                for _ in range(iterations):
+                    writer.write(b"unix")
+                    await writer.drain()
+                    assert await reader.readexactly(4) == b"unix"
+            finally:
+                writer.close()
+                await writer.wait_closed()
+        finally:
+            server.close()
+            await server.wait_closed()
+
+
+async def signals(iterations: int) -> None:
+    loop = asyncio.get_running_loop()
+    received = asyncio.Event()
+    # Registration is dispatched to rsloop's runtime thread. SIGWINCH is
+    # harmless before installation; retry delivery until that thread is ready.
+    loop.add_signal_handler(signal.SIGWINCH, received.set)
+    try:
+        for _ in range(iterations):
+            received.clear()
+            while not received.is_set():
+                os.kill(os.getpid(), signal.SIGWINCH)
+                try:
+                    await asyncio.wait_for(received.wait(), 0.05)
+                except asyncio.TimeoutError:
+                    pass
+    finally:
+        loop.remove_signal_handler(signal.SIGWINCH)
+
+
+async def cancellation(iterations: int) -> None:
+    async def waiting():
+        await asyncio.sleep(60)
+
+    for _ in range(iterations):
+        task = asyncio.create_task(waiting())
+        await asyncio.sleep(0)
+        task.cancel()
+        try:
+            await task
+        except asyncio.CancelledError:
+            pass
+        else:
+            raise AssertionError("cancellation was lost")
+
+
+RUNNERS = {
+    "threadsafe": threadsafe,
+    "executor_dns": executor_dns,
+    "udp": udp,
+    "socket_ops": socket_ops,
+    "subprocess_pipes": subprocess_pipes,
+    "unix_streams": unix_streams,
+    "signals": signals,
+    "cancellation": cancellation,
+}

benches/scheduler_workloads.py

@@ -15,8 +15,14 @@ async def bench_timers(
     """Include scheduling, expiration/cancellation, and handle release."""
     if iterations <= 0 or batch_size <= 0:
         raise ValueError("iterations and batch_size must be positive")
-    if mode not in {"timers_retained", "timers_discarded", "timers_cancelled"}:
+    if mode not in {
+        "timers_retained",
+        "timers_discarded",
+        "timers_cancelled",
+        "timers_mixed",
+    }:
         raise ValueError(f"unknown timer mode: {mode}")
+    cancelled = mode in {"timers_cancelled", "timers_mixed"}
     loop = asyncio.get_running_loop()
     started = time.perf_counter()
     outstanding = iterations
@@ -32,20 +38,28 @@ def fire(completion: asyncio.Future[None] = done) -> None:
                 completion.set_result(None)
 
         handles = []
-        for _ in range(count):
+        for index in range(count):
             if mode == "timers_discarded":
                 loop.call_later(0, fire)
             else:
-                handles.append(loop.call_later(0, fire))
-        if mode == "timers_cancelled":
+                # Deterministic scattered future deadlines exercise the heap
+                # independently of the zero-delay FIFO. Cancel before expiry.
+                delay = (
+                    60 + ((index * 15319) % 2048) / 1024
+                    if mode == "timers_mixed"
+                    else 0
+                )
+                handles.append(loop.call_later(delay, fire))
+        if cancelled:
+            handle = None
             for handle in handles:
                 handle.cancel()
             del handle
             # An expiring timer, rather than call_soon, waits for this batch's
             # cancelled deadlines to be processed as well.
             loop.call_later(0, done.set_result, None)
         await done
-        assert remaining == (count if mode == "timers_cancelled" else 0)
+        assert remaining == (count if cancelled else 0)
         handles.clear()
         outstanding -= count
     return ChildResult(

docs/development.md

@@ -19,6 +19,10 @@ Quick Rust check:
 cargo check
 ```
 
+For function timing, async poll data, allocation profiles, and a workload
+coverage inventory, see [Hotpath Profiling](hotpath-rs-profile.md). Profiling
+hooks are opt-in and absent from normal builds.
+
 The nightly pin supplies the allocator API merged in
 [rust-lang/rust#156882](https://github.com/rust-lang/rust/pull/156882). The opt-in
 `scheduler-batch-cache` Cargo feature uses the stabilized `Allocator` and

docs/hotpath-lab.md

@@ -101,13 +101,18 @@ This harness does not apply model output or create commits automatically.
   workloads; holdout is the default comparison suite.
 
 Additional opt-in scheduler probes are available through `--workloads`:
-`timers_retained`, `timers_discarded`, `timers_cancelled`, and
+`timers_retained`, `timers_discarded`, `timers_cancelled`, `timers_mixed`, and
 `tcp_connect_churn`. Timer probes schedule zero-delay timers in batches and
 include dispatch or cancellation and handle release. Retained handles remain
 live until their batch drains; discarded handles can reuse a freelist while
 callbacks remain pending. Training uses 1,000,000 timers in batches of 10,000;
 holdout uses 1,200,000 in batches of 777.
 
+`timers_mixed` instead schedules deterministic, scattered deadlines 60–62 seconds
+ahead, cancels them, and waits for their queue entries to drain. It checks heap
+costs independently of the zero-delay timer path; none of these callbacks should
+execute. Include it when changing timer data structures.
+
 The connection probe uses explicit stdlib stream protocols, so native stream
 wrappers cannot bypass generic TCP-reader startup. It creates a connection,
 echoes one payload, and closes it before the next connection. Training uses