81 lines
3.3 KiB
Python
Executable File
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()
|