Files
RustProxy/tools/checks/high_traffic_memory_model.py
T
2026-10-11 22:12:27 +02:00

81 lines
3.3 KiB
Python
Executable File

#!/usr/bin/env python3
"""Offline high-cardinality invalidation replay + HTTP load-driver smoke test.
The replay models bookkeeping cardinality, not a running RustProxy heap. The HTTP
requests target a Python test peer solely to validate the staging load harness.
"""
from __future__ import annotations
import argparse
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import os
from pathlib import Path
import sys
import threading
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "dev"))
from memory_load import run_load # noqa: E402
def purge_replay(operations: int = 300_000, windows: int = 6) -> list[tuple[int, int]]:
old_generations = {}
new_generations = {}
counts = []
assert operations % windows == 0
for i in range(operations):
key = i.to_bytes(32, "big")
# Legacy: each already-removed key keeps a generation slot forever.
old_generations[key] = 1
# Same-version fix: the global generation barrier cancels all older
# tickets, so only the current purge's per-key metadata is needed.
new_generations.clear()
new_generations[key] = 1
if (i + 1) % (operations // windows) == 0:
counts.append((len(old_generations), len(new_generations)))
assert counts == [(operations // windows * i, 1) for i in range(1, windows + 1)]
return counts
class TestPeer(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
def do_GET(self) -> None:
body = b"HTTP-load-driver-ok"
self.send_response(200)
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def log_message(self, *_: object) -> None:
pass
def main() -> None:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--http-requests", type=int, default=2_000, help="HTTP requests against the local Python test peer (default: 2000)")
args = parser.parse_args()
if not 1 <= args.http_requests <= 1_000_000:
parser.error("--http-requests must be 1..1000000")
checkpoints = purge_replay()
print("PASS: 300,000 simulated distinct PURGEs; cumulative vs reset per-key generation slots:", checkpoints)
peer = ThreadingHTTPServer(("127.0.0.1", 0), TestPeer)
runner = threading.Thread(target=peer.serve_forever, daemon=True)
runner.start()
try:
url = f"http://127.0.0.1:{peer.server_port}/"
report = run_load(url=url, requests=args.http_requests, concurrency=24, unique_paths=min(1000, args.http_requests),
timeout=5, pid=os.getpid(), sample_interval=0.05, cooldown=0.15)
assert report["requests_sent"] == args.http_requests, report
assert report["statuses"] == {"200": args.http_requests}, report
assert report["response_bytes_read"] == args.http_requests * len(b"HTTP-load-driver-ok"), report
assert report["rss_peak_mib"] is not None, report
print(f"PASS: load tool generated {args.http_requests:,} real HTTP requests with 24 connections and {min(1000, args.http_requests):,} paths against a Python test peer")
print("NOTE: peer/load-tool RSS is NOT RustProxy RSS; live RustProxy profiling remains mandatory")
finally:
peer.shutdown()
peer.server_close()
runner.join(timeout=5)
if __name__ == "__main__":
main()