diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 000000000..c102f08d7 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,48 @@ +# VCS and local editor state +.git +.gitignore +.idea/ +.vscode/ +.DS_Store + +# Docker build files are not part of the source tree copied into the image. +Dockerfile +.dockerignore + +# Python caches and local environments +__pycache__/ +*.py[cod] +*.pyo +.pytest_cache/ +.mypy_cache/ +.ruff_cache/ +.coverage +htmlcov/ +*venv*/ + +# Native and package build outputs +scratch/ +build/ +dist/ +*.egg-info/ +cmake-build-*/ +CMakeFiles/ +CMakeCache.txt +compile_commands.json +*.o +*.a +*.so +*.dylib +*.dSYM/ + +# Profiling, benchmark, and local log artifacts +*.profraw +*.profdata +*.profclangd +*.fdata +*.bolt +perf.data* +*.log + +# Misc +assets/ diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 000000000..7b7099379 --- /dev/null +++ b/Dockerfile @@ -0,0 +1,151 @@ +FROM ubuntu:24.04 + +SHELL ["/bin/bash", "-exo", "pipefail", "-c"] + +ARG DEBIAN_FRONTEND=noninteractive + +WORKDIR /root + +ENV CINDER_ROOT=/root/cinder +ENV CINDERX_ROOT=/root/cinderx +ENV CINDERX_VENV=/root/cinderx/.venv-mp312 +ENV CINDERX_BUNDLE_ARCHIVE=/root/cinderx/scratch/cinderx-bundled/libcinderx-bundled.a +ENV NETWORKBENCH_JITLIST=/root/cinderx/cinderx/benchmarks/networkbench/networkbench.jitlist.txt +ENV LLVM_VERSION=21 +ENV LLVM_ROOT=/usr/lib/llvm-$LLVM_VERSION +ENV PATH=$LLVM_ROOT/bin:$PATH +ENV PROFILE_TASK="$CINDERX_ROOT/cinderx/benchmarks/networkbench/run_server_client.py 1000" + +# Install the packages that used to be baked into ubuntu/cpython-build-benchmark. +RUN apt-get update; \ + apt-get install -yq --no-install-recommends \ + build-essential \ + ca-certificates \ + cmake \ + gdb \ + git \ + gnupg \ + linux-tools-generic \ + libbz2-dev \ + libffi-dev \ + libgdbm-compat-dev \ + libgdbm-dev \ + liblzma-dev \ + libncurses5-dev \ + libreadline-dev \ + libsqlite3-dev \ + libssl-dev \ + libzstd-dev \ + lsb-release \ + pkg-config \ + python3 \ + python3-dev \ + python3-full \ + python3-venv \ + software-properties-common \ + tk-dev \ + uuid-dev \ + wget \ + xz-utils \ + zlib1g-dev; \ + wget -O /tmp/llvm.sh https://apt.llvm.org/llvm.sh; \ + chmod +x /tmp/llvm.sh; \ + /tmp/llvm.sh "$LLVM_VERSION" all; \ + apt-get install -yq --no-install-recommends \ + "bolt-$LLVM_VERSION" \ + "libbolt-$LLVM_VERSION-dev"; \ + LLVM_BOLT="$(command -v llvm-bolt || command -v "llvm-bolt-$LLVM_VERSION")"; \ + MERGE_FDATA="$(command -v merge-fdata || command -v "merge-fdata-$LLVM_VERSION")"; \ + LLVM_PROFDATA="$(command -v llvm-profdata || command -v "llvm-profdata-$LLVM_VERSION")"; \ + PERF="$(command -v perf || find /usr/lib/linux-tools-* -type f -name perf -print -quit)"; \ + test -n "$PERF"; \ + ln -sf "$LLVM_BOLT" /usr/local/bin/llvm-bolt; \ + ln -sf "$MERGE_FDATA" /usr/local/bin/merge-fdata; \ + ln -sf "$LLVM_PROFDATA" /usr/local/bin/llvm-profdata; \ + ln -sf "$PERF" /usr/local/bin/perf; \ + command -v cmake; \ + command -v perf; \ + command -v llvm-bolt; \ + command -v merge-fdata; \ + command -v llvm-profdata; \ + command -v llvm-ar; \ + command -v llvm-ranlib; \ + llvm-bolt --version; \ + llvm-profdata --version; \ + rm -rf /var/lib/apt/lists/* /tmp/llvm.sh + +RUN git clone -b meta/3.12 https://github.com/facebookincubator/cinder.git + +WORKDIR $CINDER_ROOT + +# Build a bootstrap interpreter first. CinderX needs a working meta/3.12 Python +# to run its own PGO build before we can link those optimized objects into the +# final CPython executable. +RUN CC=clang CXX=clang++ ./configure --prefix="$CINDER_ROOT/mp312" +RUN make -j +RUN make install + +WORKDIR /root + +WORKDIR $CINDERX_ROOT + +COPY . "$CINDERX_ROOT/" + +RUN test -f "$NETWORKBENCH_JITLIST" + +RUN "$CINDER_ROOT/mp312/bin/python3" -m venv "$CINDERX_VENV" + +RUN "$CINDERX_VENV/bin/python" -m pip install --upgrade pip setuptools + +# Install CinderX with its PGO flow enabled. The local checkout supplies the +# profile-task override, so the PGO workload is networkbench instead of the +# default CPython test-suite task. +RUN PYTHONJITLISTFILE="$NETWORKBENCH_JITLIST" \ + CINDERX_ENABLE_PGO=1 \ + CINDERX_ENABLE_LTO=1 \ + CINDERX_PGO_PROFILE_TASK="$PROFILE_TASK" \ + CC=clang \ + CXX=clang++ \ + "$CINDERX_VENV/bin/pip" -vvv install --ignore-requires-python --no-build-isolation --no-clean . + +# CMake's setuptools build produces a shared _cinderx extension. For the final +# benchmark image, fold the PGO/LTO object files from that build into one static +# archive and link it as a CPython built-in module in the final optimized +# interpreter executable. +RUN CINDERX_BUILD_DIR="$(find "$CINDERX_ROOT/scratch" -maxdepth 1 -type d -name 'temp.*' -print -quit)"; \ + test -n "$CINDERX_BUILD_DIR"; \ + mkdir -p "$(dirname "$CINDERX_BUNDLE_ARCHIVE")"; \ + find "$CINDERX_BUILD_DIR" -path '*/CMakeFiles/*.dir/*' -name '*.o' -print0 \ + | sort -z \ + | xargs -0 llvm-ar rcs "$CINDERX_BUNDLE_ARCHIVE"; \ + llvm-ranlib "$CINDERX_BUNDLE_ARCHIVE"; \ + llvm-ar t "$CINDERX_BUNDLE_ARCHIVE" | grep '_cinderx.cpp.o' > /dev/null + +WORKDIR $CINDER_ROOT + +RUN make distclean + +RUN printf '%s\n' \ + '*static*' \ + "_cinderx -Wl,--whole-archive $CINDERX_BUNDLE_ARCHIVE -Wl,--no-whole-archive -lstdc++ -lz -ldl -lpthread -lm" \ + > Modules/Setup.local; \ + cat Modules/Setup.local + +# The final binary includes CinderX, so keep BOLT's core layout passes while +# avoiding LLVM 21 AArch64 apply passes that crash or exceed the VM memory limit. +RUN CC=clang \ + CXX=clang++ \ + LDFLAGS="-fuse-ld=lld -flto" \ + BOLT_APPLY_FLAGS="-skip-funcs=_PyEval_EvalFrameDefault,sre_ucs1_match/1,sre_ucs2_match/1,sre_ucs4_match/1 -reorder-blocks=ext-tsp -reorder-functions=cdsort -split-functions -reorder-functions-use-hot-size -peepholes=none -use-gnu-stack" \ + ./configure --prefix="$CINDER_ROOT/mp312" --enable-optimizations --enable-bolt +RUN PYTHONPATH="$CINDERX_ROOT/cinderx/PythonLib" \ + PYTHONJITLISTFILE="$NETWORKBENCH_JITLIST" \ + make -j +RUN make install + +WORKDIR $CINDERX_ROOT + +RUN find "$CINDERX_VENV" -name '_cinderx*.so' -delete; \ + "$CINDERX_VENV/bin/python" -c 'import importlib.util; spec = importlib.util.find_spec("_cinderx"); assert spec is not None and spec.origin == "built-in", spec; import cinderx; assert cinderx.is_initialized(), cinderx.get_import_error()' + +ENTRYPOINT ["bash", "-c", "source $CINDERX_VENV/bin/activate && python cinderx/benchmarks/networkbench/run_bench.py -n 1"] diff --git a/cinderx/benchmarks/networkbench/README.md b/cinderx/benchmarks/networkbench/README.md new file mode 100644 index 000000000..36899135b --- /dev/null +++ b/cinderx/benchmarks/networkbench/README.md @@ -0,0 +1,62 @@ +# networkbench + +`networkbench` is a small benchmark meant to reproduce the shape of a Python +webserver. It exercises request parsing, route dispatch, middleware, async I/O, +filesystem-backed state, response encoding, and CPU work inside request handlers. +(File I/O through async thread executors substitutes for database query I/O.) + +The benchmark runs a local HTTP server on `localhost:8080` and drives it with an +async client. The client uploads network matrices, then sends concurrent requests +against two main endpoints: + +- `GET /network` reads a stored matrix and returns it. +- `GET /reachable` decodes a graph payload and computes whether two nodes are + reachable. + + +Performance is measured in the client as number of requests/second, after the +upload phase. + +## Running + +Run one server/client benchmark pass: + +```bash +cd cinderx/benchmarks/networkbench +python run_server_client.py 10000 +``` + +The numeric argument is the number of client requests. The client prints +`Average requests per second` after the run. + +Run the comparison harness: + +```bash +python run_bench.py 10000 -n 5 +``` + +`run_bench.py` compares: + +- `cinderx_jitlist`: runs with `networkbench.jitlist.txt`. +- `cinderx_disable`: runs with `CINDERX_DISABLE=1`. + +## Useful Knobs + +These environment variables tune the workload: + +- `SERVER_PROCESS_COUNT`: number of worker processes, default `8`. +- `CLIENT_MAX_INFLIGHT_REQUESTS`: client concurrency limit, default `16`. +- `NETWORK_MATRIX_COUNT`: matrices uploaded before the timed run, default `16`. +- `NETWORK_GET_PERCENT`: percentage of timed requests sent to `/network`, + default `90`. +- `NETWORK_STORAGE_DIR`: directory used for stored matrix files. + +## JIT List + +Regenerate the JIT list from a debug run with: + +```bash +python generate_networkbench_jitlist.py 10000 +``` + +This rewrites `networkbench.jitlist.txt` by default. diff --git a/cinderx/benchmarks/networkbench/client.py b/cinderx/benchmarks/networkbench/client.py new file mode 100644 index 000000000..6b954256d --- /dev/null +++ b/cinderx/benchmarks/networkbench/client.py @@ -0,0 +1,240 @@ +import argparse +import asyncio +from collections.abc import Mapping, Sequence +import random +import time +from typing import TypeAlias, TypedDict + +import config +import network_data +import matrix_codec + + +HEADER_ENCODING = "iso-8859-1" +HTTPHeaderValue: TypeAlias = str | int +HTTPHeaders: TypeAlias = Mapping[str, HTTPHeaderValue] +HTTPResponse: TypeAlias = tuple[int, bytes] + + +class ReachabilityPayload(TypedDict): + source: int + destination: int + graph: list[list[int]] + + +def encode_http_request( + method: str, + path: str, + headers: HTTPHeaders, + body: bytes = b"", +) -> bytes: + encoded_headers = [ + f"{method} {path} HTTP/1.1\r\n", + f"Host: {config.HOST}:{config.PORT}\r\n", + "Connection: close\r\n", + ] + for name, value in headers.items(): + encoded_headers.append(f"{name}: {value}\r\n") + encoded_headers.append("\r\n") + return "".join(encoded_headers).encode("ascii") + body + + +async def read_http_response(reader: asyncio.StreamReader) -> HTTPResponse: + status_line = await reader.readline() + if not status_line: + raise RuntimeError("Empty response") + + parts = status_line.decode(HEADER_ENCODING).split(" ", 2) + if len(parts) < 2: + raise RuntimeError("Invalid response status line") + status = int(parts[1]) + + headers: dict[str, str] = {} + while True: + line = await reader.readline() + if line in (b"\r\n", b"\n", b""): + break + name, value = line.decode(HEADER_ENCODING).split(":", 1) + headers[name.strip().lower()] = value.strip() + + content_length = headers.get("content-length") + if content_length is None: + body = await reader.read() + else: + body = await reader.readexactly(int(content_length)) + return status, body + + +async def make_http_request( + method: str, + path: str, + headers: HTTPHeaders | None = None, + body: bytes = b"", +) -> HTTPResponse: + reader, writer = await asyncio.open_connection(config.HOST, config.PORT) + try: + writer.write(encode_http_request(method, path, headers or {}, body)) + await writer.drain() + return await read_http_response(reader) + finally: + writer.close() + await writer.wait_closed() + + +async def wait_for_server() -> None: + while True: + try: + status, _body = await make_http_request("GET", config.STATUS_PATH) + except Exception: + await asyncio.sleep(0.01) + continue + if status == 200: + return + + +def make_reachability_payload() -> ReachabilityPayload: + return { + "source": network_data.SOURCE_NODE, + "destination": network_data.DESTINATION_NODE, + "graph": network_data.REACHABILITY_MATRIX, + } + + +def encode_reachability_body(payload: ReachabilityPayload) -> bytes: + return matrix_codec.encode_reachability_request( + payload["graph"], + payload["source"], + payload["destination"], + ) + + +_cached_reachability_body = encode_reachability_body(make_reachability_payload()) + + +def reachability_headers(body: bytes) -> dict[str, HTTPHeaderValue]: + return { + "Content-Type": config.REACHABILITY_CONTENT_TYPE, + "Content-Length": len(body), + } + + +def network_post_headers( + network_id: int, + matrix_size: int, + body: bytes, +) -> dict[str, HTTPHeaderValue]: + return { + "Content-Type": config.NETWORK_CONTENT_TYPE, + "Content-Length": len(body), + "X-Network-Id": str(network_id), + "X-Network-Size": str(matrix_size), + } + + +def network_get_headers(network_id: int) -> dict[str, HTTPHeaderValue]: + return {"X-Network-Id": str(network_id)} + + +async def get_reachability_response() -> HTTPResponse: + body = _cached_reachability_body + return await make_http_request( + "GET", + config.REACHABLE_PATH, + headers=reachability_headers(body), + body=body, + ) + + +async def post_network_matrix(network_id: int) -> None: + matrix = network_data.build_reachability_matrix() + body = matrix_codec.encode_square_matrix(matrix) + status, _body = await make_http_request( + "POST", + config.NETWORK_PATH, + headers=network_post_headers(network_id, len(matrix), body), + body=body, + ) + if status != 200: + raise RuntimeError(f"POST /network failed: {status}") + + +async def upload_network_matrices() -> None: + for network_id in range(config.NETWORK_MATRIX_COUNT): + await post_network_matrix(network_id) + + +async def make_requests(request_count: int) -> None: + semaphore = asyncio.Semaphore(config.CLIENT_MAX_INFLIGHT_REQUESTS) + + async def make_bounded_request(request_index: int) -> None: + async with semaphore: + await make_request(request_index) + + await asyncio.gather( + *(make_bounded_request(request_index) for request_index in range(request_count)) + ) + + +async def make_request(request_index: int) -> None: + if is_network_get_request(request_index): + await make_network_request(request_index) + else: + await make_reachability_request() + + +def is_network_get_request(_request_index: int) -> bool: + return random.randrange(100) < config.NETWORK_GET_PERCENT + + +async def make_network_request(request_index: int) -> None: + network_id = request_index % config.NETWORK_MATRIX_COUNT + status, _body = await make_http_request( + "GET", + config.NETWORK_PATH, + headers=network_get_headers(network_id), + ) + if status != 200: + raise RuntimeError(f"GET /network failed: {status}") + + +async def make_reachability_request() -> None: + status, _body = await get_reachability_response() + if status != 200: + raise RuntimeError(f"GET /reachable failed: {status}") + + +def positive_int(value: str) -> int: + parsed = int(value) + if parsed < 1: + raise argparse.ArgumentTypeError("must be greater than 0") + return parsed + + +def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument( + "request_count", + nargs="?", + type=positive_int, + default=1, + help="number of benchmark requests to send", + ) + return parser.parse_args(argv) + + +async def run_benchmark(args: argparse.Namespace) -> None: + await wait_for_server() + await upload_network_matrices() + started = time.perf_counter() + await make_requests(args.request_count) + elapsed = time.perf_counter() - started + print(f"Average requests per second: {args.request_count / elapsed:.2f}") + + +def main(argv: Sequence[str] | None = None) -> None: + args = parse_args(argv) + asyncio.run(run_benchmark(args)) + + +if __name__ == "__main__": + main() diff --git a/cinderx/benchmarks/networkbench/config.py b/cinderx/benchmarks/networkbench/config.py new file mode 100644 index 000000000..085cdb98b --- /dev/null +++ b/cinderx/benchmarks/networkbench/config.py @@ -0,0 +1,20 @@ +import os +import tempfile + + +SERVER_PROCESS_COUNT = int(os.getenv("SERVER_PROCESS_COUNT", "8")) +CLIENT_MAX_INFLIGHT_REQUESTS = int(os.getenv("CLIENT_MAX_INFLIGHT_REQUESTS", "16")) +HOST, PORT = "localhost", 8080 +STATUS_PATH = "/status" +REACHABLE_PATH = "/reachable" +NETWORK_PATH = "/network" +JSON_CONTENT_TYPE = "application/json" +REACHABILITY_CONTENT_TYPE = "application/vnd.networkbench.reachability" +NETWORK_CONTENT_TYPE = "application/vnd.networkbench.matrix" +MAX_REQUEST_BODY_BYTES = 400_000_000 +NETWORK_MATRIX_COUNT = int(os.getenv("NETWORK_MATRIX_COUNT", "16")) +NETWORK_GET_PERCENT = int(os.getenv("NETWORK_GET_PERCENT", "90")) +NETWORK_STORAGE_DIR = os.getenv( + "NETWORK_STORAGE_DIR", + os.path.join(tempfile.gettempdir(), "networkbench-matrices"), +) diff --git a/cinderx/benchmarks/networkbench/generate_networkbench_jitlist.py b/cinderx/benchmarks/networkbench/generate_networkbench_jitlist.py new file mode 100644 index 000000000..6b69a7bfe --- /dev/null +++ b/cinderx/benchmarks/networkbench/generate_networkbench_jitlist.py @@ -0,0 +1,117 @@ +#!/usr/bin/env python3 +import argparse +import os +import re +import subprocess +import sys +import tempfile +from pathlib import Path + + +SCRIPT_DIR = Path(__file__).resolve().parent +REPO_ROOT = SCRIPT_DIR.parents[2] +RUNNER = Path("cinderx/benchmarks/networkbench/run_server_client.py") +DEFAULT_OUTPUT = Path(__file__).with_name("networkbench.jitlist.txt") +DEFAULT_REQUEST_COUNT = 10000 +COMPILED_FUNC_RE = re.compile(r"^JIT: .* -- Finished compiling (\S+) in .*$") + + +def parse_compiled_functions(log_output: str) -> list[str]: + funcs: set[str] = set() + for line in log_output.splitlines(): + match = COMPILED_FUNC_RE.search(line) + if match is not None: + funcs.add(match.group(1)) + return sorted(funcs) + + +def read_jit_logs(log_dir: Path) -> str: + chunks: list[str] = [] + for path in sorted(log_dir.glob("jit.*.log")): + chunks.append(path.read_text(encoding="ascii", errors="replace")) + return "".join(chunks) + + +def run_networkbench(python: str, request_count: int) -> str: + env = dict(os.environ) + command = [ + python, + str(RUNNER), + str(request_count), + ] + with tempfile.TemporaryDirectory(prefix="networkbench-jitlog-") as log_dir_str: + log_dir = Path(log_dir_str) + # Keep JIT logs out of the shared benchmark stdout/stderr stream. + env.update( + { + "PYTHONJITDEBUG": "1", + "CINDERX_JIT_LOG_FILE": str(log_dir / "jit.{pid}.log"), + } + ) + proc = subprocess.run( + command, + cwd=REPO_ROOT, + env=env, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + encoding="ascii", + errors="replace", + ) + log_output = read_jit_logs(log_dir) + if proc.returncode != 0: + sys.stderr.write(proc.stdout) + raise SystemExit(f"{' '.join(command)} failed with exit code {proc.returncode}") + return log_output or proc.stdout + + +def write_jitlist(functions: list[str], output_path: Path) -> None: + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text("".join(f"{func}\n" for func in functions), encoding="ascii") + + +def positive_int(value: str) -> int: + parsed = int(value) + if parsed < 1: + raise argparse.ArgumentTypeError("must be greater than 0") + return parsed + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Regenerate the networkbench jitlist from a JIT debug run." + ) + parser.add_argument( + "request_count", + nargs="?", + default=DEFAULT_REQUEST_COUNT, + type=positive_int, + help=f"number of networkbench requests to run (default: {DEFAULT_REQUEST_COUNT})", + ) + parser.add_argument( + "--python", + default="python", + help="Python executable to use for the benchmark run (default: python)", + ) + parser.add_argument( + "-o", + "--output", + default=DEFAULT_OUTPUT, + type=Path, + help=f"jitlist file to write (default: {DEFAULT_OUTPUT})", + ) + return parser.parse_args() + + +def main() -> None: + args = parse_args() + log_output = run_networkbench(args.python, args.request_count) + functions = parse_compiled_functions(log_output) + if not functions: + raise SystemExit("No compiled functions found in networkbench output") + write_jitlist(functions, args.output) + print(f"Wrote {len(functions)} functions to {args.output}") + + +if __name__ == "__main__": + main() diff --git a/cinderx/benchmarks/networkbench/matrix_codec.py b/cinderx/benchmarks/networkbench/matrix_codec.py new file mode 100644 index 000000000..c74a40259 --- /dev/null +++ b/cinderx/benchmarks/networkbench/matrix_codec.py @@ -0,0 +1,104 @@ +from typing import TypedDict + + +Matrix = list[list[int]] + + +class ReachabilityRequest(TypedDict): + source: int + destination: int + graph: Matrix + + +MAGIC = b"NBM1" +UINT32_BYTES = 4 +HEADER_BYTES = len(MAGIC) + UINT32_BYTES * 3 +MAX_UINT32 = (1 << 32) - 1 + + +def _append_uint32(encoded: bytearray, value: int) -> None: + if not isinstance(value, int): + raise ValueError("integer field is not an int") + if value < 0 or value > MAX_UINT32: + raise ValueError("integer field is out of range") + + encoded.append((value >> 24) & 0xFF) + encoded.append((value >> 16) & 0xFF) + encoded.append((value >> 8) & 0xFF) + encoded.append(value & 0xFF) + + +def _read_uint32(encoded: bytes, offset: int) -> int: + return ( + (encoded[offset] << 24) + | (encoded[offset + 1] << 16) + | (encoded[offset + 2] << 8) + | encoded[offset + 3] + ) + + +def _validate_square_matrix(matrix: Matrix) -> int: + size = len(matrix) + for row in matrix: + if len(row) != size: + raise ValueError("matrix is not square") + return size + + +def encode_square_matrix(matrix: Matrix) -> bytes: + size = _validate_square_matrix(matrix) + encoded = bytearray(size * size) + offset = 0 + for row in matrix: + for value in row: + if value != 0 and value != 1: + raise ValueError("matrix values must be 0 or 1") + encoded[offset] = value + offset += 1 + return bytes(encoded) + + +def decode_square_matrix(encoded: bytes, size: int) -> Matrix: + expected_length = size * size + if len(encoded) != expected_length: + raise ValueError("encoded matrix has the wrong length") + + matrix = [] + offset = 0 + for _ in range(size): + next_offset = offset + size + matrix.append(list(encoded[offset:next_offset])) + offset = next_offset + return matrix + + +def encode_reachability_request( + matrix: Matrix, + source: int, + destination: int, +) -> bytes: + size = _validate_square_matrix(matrix) + encoded = bytearray() + encoded.extend(MAGIC) + _append_uint32(encoded, size) + _append_uint32(encoded, source) + _append_uint32(encoded, destination) + encoded.extend(encode_square_matrix(matrix)) + return bytes(encoded) + + +def decode_reachability_request(encoded: bytes) -> ReachabilityRequest: + if len(encoded) < HEADER_BYTES: + raise ValueError("request body is too short") + if encoded[: len(MAGIC)] != MAGIC: + raise ValueError("invalid request body magic") + + size = _read_uint32(encoded, len(MAGIC)) + source = _read_uint32(encoded, len(MAGIC) + UINT32_BYTES) + destination = _read_uint32(encoded, len(MAGIC) + UINT32_BYTES * 2) + matrix = decode_square_matrix(encoded[HEADER_BYTES:], size) + return { + "source": source, + "destination": destination, + "graph": matrix, + } diff --git a/cinderx/benchmarks/networkbench/network_data.py b/cinderx/benchmarks/networkbench/network_data.py new file mode 100644 index 000000000..b03e08dbd --- /dev/null +++ b/cinderx/benchmarks/networkbench/network_data.py @@ -0,0 +1,22 @@ +NODE_COUNT = 4_000 +SOURCE_NODE = 0 +DESTINATION_NODE = NODE_COUNT - 1 +FORWARD_EDGE_COUNT = 200 +LONG_EDGE_COUNT = 25 +LONG_EDGE_STRIDE = 37 + + +def build_reachability_matrix() -> list[list[int]]: + matrix = [] + for source in range(NODE_COUNT): + row = [0] * NODE_COUNT + for offset in range(1, FORWARD_EDGE_COUNT + 1): + row[(source + offset) % NODE_COUNT] = 1 + for offset in range(1, LONG_EDGE_COUNT + 1): + row[(source + offset * LONG_EDGE_STRIDE) % NODE_COUNT] = 1 + row[source] = 0 + matrix.append(row) + return matrix + + +REACHABILITY_MATRIX = build_reachability_matrix() diff --git a/cinderx/benchmarks/networkbench/network_lib.py b/cinderx/benchmarks/networkbench/network_lib.py new file mode 100644 index 000000000..079035736 --- /dev/null +++ b/cinderx/benchmarks/networkbench/network_lib.py @@ -0,0 +1,38 @@ +from very_simple_queue import VerySimpleQueue + + +def are_reachable( + graph: list[list[int]], + source: int, + destination: int, +) -> tuple[bool, list[int]]: + queue = VerySimpleQueue() + explored = [False for _ in range(len(graph))] + reached_from = [-1 for _ in range(len(graph))] + explored[source] = True + queue.put(source) + while not queue.empty(): + v = queue.get() + if v == destination: + return True, produce_path(reached_from, source, destination) + for w, reachable in enumerate(graph[v]): + if not reachable: + continue + if not explored[w]: + explored[w] = True + queue.put(w) + assert reached_from[w] == -1 + reached_from[w] = v + return False, [] + + +def produce_path( + reached_from: list[int], + source: int, + destination: int, +) -> list[int]: + reverse_path = [destination] + while (latest := reverse_path[-1]) != source: + reverse_path.append(reached_from[latest]) + assert reverse_path[-1] == source + return list(reversed(reverse_path)) diff --git a/cinderx/benchmarks/networkbench/networkbench.jitlist.txt b/cinderx/benchmarks/networkbench/networkbench.jitlist.txt new file mode 100644 index 000000000..6f8522a45 --- /dev/null +++ b/cinderx/benchmarks/networkbench/networkbench.jitlist.txt @@ -0,0 +1,181 @@ +_weakrefset:WeakSet.__len__ +_weakrefset:WeakSet.add +_weakrefset:WeakSet.discard +asyncio.base_events:BaseEventLoop._add_callback +asyncio.base_events:BaseEventLoop._call_soon +asyncio.base_events:BaseEventLoop._check_closed +asyncio.base_events:BaseEventLoop._check_default_executor +asyncio.base_events:BaseEventLoop._run_once +asyncio.base_events:BaseEventLoop.call_soon +asyncio.base_events:BaseEventLoop.call_soon_threadsafe +asyncio.base_events:BaseEventLoop.create_future +asyncio.base_events:BaseEventLoop.create_task +asyncio.base_events:BaseEventLoop.get_debug +asyncio.base_events:BaseEventLoop.is_closed +asyncio.base_events:BaseEventLoop.run_in_executor +asyncio.base_events:BaseEventLoop.time +asyncio.base_events:Server._attach +asyncio.base_events:Server._detach +asyncio.base_events:_set_nodelay +asyncio.base_futures:isfuture +asyncio.events:Handle.__init__ +asyncio.events:Handle._run +asyncio.events:Handle.cancel +asyncio.events:_ThreadSafeHandle.__init__ +asyncio.events:_ThreadSafeHandle._run +asyncio.futures:_chain_future +asyncio.futures:_chain_future.._call_check_cancel +asyncio.futures:_chain_future.._call_set_state +asyncio.futures:_chain_future.._set_state +asyncio.futures:_copy_future_state +asyncio.futures:_get_loop +asyncio.futures:_set_result_unless_cancelled +asyncio.futures:wrap_future +asyncio.protocols:BaseProtocol.connection_lost +asyncio.protocols:BaseProtocol.pause_writing +asyncio.protocols:BaseProtocol.resume_writing +asyncio.selector_events:BaseSelectorEventLoop._accept_connection +asyncio.selector_events:BaseSelectorEventLoop._accept_connection2 +asyncio.selector_events:BaseSelectorEventLoop._add_reader +asyncio.selector_events:BaseSelectorEventLoop._add_writer +asyncio.selector_events:BaseSelectorEventLoop._ensure_fd_no_transport +asyncio.selector_events:BaseSelectorEventLoop._make_socket_transport +asyncio.selector_events:BaseSelectorEventLoop._process_events +asyncio.selector_events:BaseSelectorEventLoop._read_from_self +asyncio.selector_events:BaseSelectorEventLoop._remove_reader +asyncio.selector_events:BaseSelectorEventLoop._remove_writer +asyncio.selector_events:BaseSelectorEventLoop._write_to_self +asyncio.selector_events:_SelectorSocketTransport.__init__ +asyncio.selector_events:_SelectorSocketTransport._adjust_leftover_buffer +asyncio.selector_events:_SelectorSocketTransport._call_connection_lost +asyncio.selector_events:_SelectorSocketTransport._get_sendmsg_buffer +asyncio.selector_events:_SelectorSocketTransport._read_ready +asyncio.selector_events:_SelectorSocketTransport._read_ready__data_received +asyncio.selector_events:_SelectorSocketTransport._write_sendmsg +asyncio.selector_events:_SelectorSocketTransport.close +asyncio.selector_events:_SelectorSocketTransport.set_protocol +asyncio.selector_events:_SelectorSocketTransport.write +asyncio.selector_events:_SelectorTransport.__del__ +asyncio.selector_events:_SelectorTransport.__init__ +asyncio.selector_events:_SelectorTransport._add_reader +asyncio.selector_events:_SelectorTransport._call_connection_lost +asyncio.selector_events:_SelectorTransport.close +asyncio.selector_events:_SelectorTransport.get_write_buffer_size +asyncio.selector_events:_SelectorTransport.is_closing +asyncio.selector_events:_SelectorTransport.is_reading +asyncio.selector_events:_SelectorTransport.set_protocol +asyncio.tasks:create_task +asyncio.transports:BaseTransport.__init__ +asyncio.transports:BaseTransport.get_extra_info +asyncio.transports:_FlowControlMixin.__init__ +asyncio.transports:_FlowControlMixin._maybe_pause_protocol +asyncio.transports:_FlowControlMixin._maybe_resume_protocol +asyncio.transports:_FlowControlMixin._set_write_buffer_limits +asyncio.trsock:TransportSocket.__init__ +asyncio.unix_events:_UnixSelectorEventLoop._process_self_data +concurrent.futures._base:Future.__get_result +concurrent.futures._base:Future.__init__ +concurrent.futures._base:Future._invoke_callbacks +concurrent.futures._base:Future.add_done_callback +concurrent.futures._base:Future.cancelled +concurrent.futures._base:Future.done +concurrent.futures._base:Future.exception +concurrent.futures._base:Future.result +concurrent.futures._base:Future.set_result +concurrent.futures._base:Future.set_running_or_notify_cancel +concurrent.futures.thread:ThreadPoolExecutor._adjust_thread_count +concurrent.futures.thread:ThreadPoolExecutor.submit +concurrent.futures.thread:WorkerContext.prepare..resolve_task +concurrent.futures.thread:WorkerContext.run +concurrent.futures.thread:_WorkItem.__init__ +concurrent.futures.thread:_WorkItem.run +enum:Enum.__new__ +enum:Enum.value +enum:EnumType.__call__ +enum:property.__get__ +namedtuple_SelectorKey:SelectorKey.__new__ +network_lib:are_reachable +network_lib:produce_path +posixpath:_get_sep +posixpath:join +selectors:KqueueSelector.register +selectors:KqueueSelector.select +selectors:KqueueSelector.unregister +selectors:_BaseSelectorImpl._fileobj_lookup +selectors:_BaseSelectorImpl.get_map +selectors:_BaseSelectorImpl.modify +selectors:_BaseSelectorImpl.register +selectors:_BaseSelectorImpl.unregister +selectors:_SelectorMapping.get +selectors:_fileobj_to_fd +simple_web_framework:BaseRoute.finalize_response +simple_web_framework:BaseRoute.prepare +simple_web_framework:BaseRoute.response_headers +simple_web_framework:BaseViewMiddleware.after_response +simple_web_framework:BaseViewMiddleware.before_request +simple_web_framework:Connection.__init__ +simple_web_framework:Connection.connection_made +simple_web_framework:Connection.data_received +simple_web_framework:Connection.dispatch_request +simple_web_framework:Connection.find_route +simple_web_framework:Connection.handle +simple_web_framework:Connection.read_request +simple_web_framework:Connection.send_error +simple_web_framework:Connection.send_response +simple_web_framework:HTTPServer.create_connection +simple_web_framework:Request.__init__ +simple_web_framework:RequestContext.__init__ +simple_web_framework:ViewStack.after_view +simple_web_framework:ViewStack.before_view +socket:_intenum_converter +socket:socket.__init__ +socket:socket._real_close +socket:socket.accept +socket:socket.close +socket:socket.family +socket:socket.type +threading:Condition.__enter__ +threading:Condition.__exit__ +threading:Condition.__init__ +threading:Condition._is_owned +threading:Condition.notify +threading:Condition.notify_all +threading:RLock +threading:Semaphore.acquire +threading:Semaphore.release +traffic_stats_static:WebTrafficStats.record_request +traffic_stats_static:WebTrafficStats.record_request_elapsed +traffic_stats_static:WebTrafficStats.record_response +traffic_stats_static:WebTrafficStats.snapshot +traffic_stats_static:WebTrafficStats.status_codes_snapshot +very_simple_queue:Item.__init__ +very_simple_queue:VerySimpleQueue.empty +very_simple_queue:VerySimpleQueue.get +very_simple_queue:VerySimpleQueue.put +views:NetworkGetRoute.encode_response +views:NetworkGetRoute.handle +views:NetworkGetRoute.prepare +views:NetworkPostRoute.handle +views:NetworkPostRoute.prepare +views:NetworkRoute.matrix_path +views:NetworkRoute.parse_int_header +views:NetworkRoute.parse_network_id +views:NetworkRoute.read_matrix +views:NetworkRoute.read_matrix_file +views:NetworkRoute.response_headers +views:PayloadAuditMiddleware.after_response +views:ReachabilityRoute.prepare +views:ReachabilityRoute.read_request +views:RequestHeadersMiddleware.after_response +views:RequestHeadersMiddleware.before_request +views:ResponseHeadersMiddleware.after_response +views:RouteMetadataMiddleware.after_response +views:RouteMetadataMiddleware.before_request +views:StatusRoute.handle +views:TimingMiddleware.after_response +views:TimingMiddleware.before_request +weakref:KeyedRef.__init__ +weakref:KeyedRef.__new__ +weakref:WeakValueDictionary.__init__..remove +weakref:WeakValueDictionary.__setitem__ +weakref:WeakValueDictionary.get diff --git a/cinderx/benchmarks/networkbench/run_bench.py b/cinderx/benchmarks/networkbench/run_bench.py new file mode 100644 index 000000000..ee0f8f9f2 --- /dev/null +++ b/cinderx/benchmarks/networkbench/run_bench.py @@ -0,0 +1,227 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import contextlib +import os +import re +import statistics +import subprocess +import sys +from dataclasses import dataclass +from pathlib import Path + + +DEFAULT_REQUEST_COUNT = 10000 +DEFAULT_RUNS = 5 +SCRIPT_DIR = Path(__file__).resolve().parent +RUNNER = "run_server_client.py" +JITLIST = "networkbench.jitlist.txt" +REQUESTS_PER_SECOND_RE = re.compile( + r"Average requests per second:\s+([0-9]+(?:\.[0-9]+)?)" +) + + +@dataclass(frozen=True) +class BenchmarkCommand: + name: str + env: dict[str, str] + + +@dataclass(frozen=True) +class BenchmarkResult: + name: str + requests_per_second: list[float] + + @property + def mean(self) -> float: + return statistics.fmean(self.requests_per_second) + + @property + def min(self) -> float: + return min(self.requests_per_second) + + @property + def stdev(self) -> float: + if len(self.requests_per_second) < 2: + return 0.0 + return statistics.stdev(self.requests_per_second) + + @property + def sorted_requests_per_second(self) -> list[float]: + return sorted(self.requests_per_second, reverse=True) + + +def positive_int(value: str) -> int: + parsed = int(value) + if parsed < 1: + raise argparse.ArgumentTypeError("must be greater than 0") + return parsed + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Compare networkbench with a JIT list against CinderX disabled." + ) + parser.add_argument( + "request_count", + nargs="?", + default=DEFAULT_REQUEST_COUNT, + type=positive_int, + help=f"number of client requests per run (default: {DEFAULT_REQUEST_COUNT})", + ) + parser.add_argument( + "-n", + "--runs", + default=DEFAULT_RUNS, + type=positive_int, + help=f"number of runs for each command (default: {DEFAULT_RUNS})", + ) + parser.add_argument( + "--python", + default="python", + help="Python executable to use for benchmark commands (default: python)", + ) + return parser.parse_args() + + +def make_commands(networkbench_dir: Path) -> list[BenchmarkCommand]: + jitlist = (networkbench_dir / JITLIST).resolve() + if not jitlist.is_file(): + raise SystemExit(f"JIT list not found: {jitlist}") + + return [ + BenchmarkCommand("cinderx_jitlist", {"PYTHONJITLISTFILE": str(jitlist)}), + BenchmarkCommand("cinderx_disable", {"CINDERX_DISABLE": "1"}), + ] + + +def parse_requests_per_second(output: str) -> float | None: + match = REQUESTS_PER_SECOND_RE.search(output) + if match is None: + return None + return float(match.group(1)) + + +def run_command( + command: BenchmarkCommand, + python: str, + request_count: int, + networkbench_dir: Path, +) -> float: + env = dict(os.environ) + env.update(command.env) + args = [python, RUNNER, str(request_count)] + + proc = subprocess.run( + args, + cwd=networkbench_dir, + env=env, + stdout=subprocess.PIPE, + stderr=subprocess.STDOUT, + text=True, + encoding="utf-8", + errors="replace", + ) + + if proc.returncode != 0: + sys.stderr.write(proc.stdout) + raise SystemExit( + f"{command.name} failed with exit code {proc.returncode}: " + f"{format_command(command, python, request_count)}" + ) + + requests_per_second = parse_requests_per_second(proc.stdout) + if requests_per_second is None: + sys.stderr.write(proc.stdout) + raise SystemExit( + f"{command.name} output did not report requests per second: " + f"{format_command(command, python, request_count)}" + ) + + return requests_per_second + + +def run_benchmark( + command: BenchmarkCommand, + python: str, + request_count: int, + runs: int, + networkbench_dir: Path, +) -> BenchmarkResult: + requests_per_second = [] + for run in range(1, runs + 1): + print(f"{command.name}: run {run}/{runs} ...", flush=True) + requests_per_second.append( + run_command(command, python, request_count, networkbench_dir) + ) + return BenchmarkResult(command.name, requests_per_second) + + +def format_command(command: BenchmarkCommand, python: str, request_count: int) -> str: + env = " ".join(f'{key}="{value}"' for key, value in command.env.items()) + return f"{env} {python} {RUNNER} {request_count}" + + +def print_summary( + results: list[BenchmarkResult], + commands: list[BenchmarkCommand], + python: str, + request_count: int, +) -> None: + print() + print("Commands:") + for command in commands: + print(f" {command.name}: {format_command(command, python, request_count)}") + + print() + print( + f"{'benchmark':<18} " + f"{'req/s mean':>12} " + f"{'req/s min':>12} " + f"{'req/s stdev':>12} " + f"runs" + ) + print("-" * 72) + for result in results: + runs = ", ".join( + f"{requests_per_second:.2f}" + for requests_per_second in result.sorted_requests_per_second + ) + print( + f"{result.name:<18} " + f"{result.mean:>12.2f} " + f"{result.min:>12.2f} " + f"{result.stdev:>12.2f} " + f"{runs}" + ) + + if len(results) == 2: + baseline, contender = results + ratio = contender.mean / baseline.mean + faster = contender.name if ratio >= 1 else baseline.name + speedup = ratio if ratio >= 1 else 1 / ratio + print() + print(f"{faster} reports {speedup:.2f}x higher mean requests/second.") + + +def main() -> None: + args = parse_args() + with contextlib.chdir(SCRIPT_DIR): + networkbench_dir = Path.cwd().resolve() + commands = make_commands(networkbench_dir) + results = [ + run_benchmark( + command, + args.python, + args.request_count, + args.runs, + networkbench_dir, + ) + for command in commands + ] + print_summary(results, commands, args.python, args.request_count) + + +if __name__ == "__main__": + main() diff --git a/cinderx/benchmarks/networkbench/run_server_client.py b/cinderx/benchmarks/networkbench/run_server_client.py new file mode 100644 index 000000000..3ecedeea7 --- /dev/null +++ b/cinderx/benchmarks/networkbench/run_server_client.py @@ -0,0 +1,70 @@ +import argparse +from collections.abc import Sequence +import subprocess +import sys +from pathlib import Path + + +SCRIPT_DIR = Path(__file__).resolve().parent +SERVER_SCRIPT = SCRIPT_DIR / "server.py" +CLIENT_SCRIPT = SCRIPT_DIR / "client.py" +SERVER_SHUTDOWN_TIMEOUT = 5 + + +def start_script(script: Path, *args: str) -> subprocess.Popen[bytes]: + return subprocess.Popen( + [sys.executable, str(script), *args], + cwd=SCRIPT_DIR, + ) + + +def stop_process(process: subprocess.Popen[bytes]) -> None: + if process.poll() is not None: + return + + process.terminate() + try: + process.wait(timeout=SERVER_SHUTDOWN_TIMEOUT) + except subprocess.TimeoutExpired: + process.kill() + process.wait() + + +def run_client_with_server(request_count: int) -> int: + server_process = start_script(SERVER_SCRIPT) + try: + client_process = start_script(CLIENT_SCRIPT, str(request_count)) + return client_process.wait() + finally: + stop_process(server_process) + + +def positive_int(value: str) -> int: + parsed = int(value) + if parsed < 1: + raise argparse.ArgumentTypeError("must be greater than 0") + return parsed + + +def parse_args(argv: Sequence[str] | None = None) -> argparse.Namespace: + parser = argparse.ArgumentParser() + parser.add_argument( + "request_count", + nargs="?", + type=positive_int, + default=1, + help="number of benchmark requests for the client to send", + ) + return parser.parse_args(argv) + + +def main(argv: Sequence[str] | None = None) -> int: + try: + args = parse_args(argv) + return run_client_with_server(args.request_count) + except KeyboardInterrupt: + return 130 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/cinderx/benchmarks/networkbench/server.py b/cinderx/benchmarks/networkbench/server.py new file mode 100644 index 000000000..ede70b3d3 --- /dev/null +++ b/cinderx/benchmarks/networkbench/server.py @@ -0,0 +1,178 @@ +import asyncio +from collections.abc import Sequence +import json +import multiprocessing +import os +import signal +import socket +import sys +import time + +try: + import cinderx.jit + HAS_CINDERX = cinderx.jit.is_enabled() +except ImportError: + HAS_CINDERX = False + +import config +import network_lib +from simple_web_framework import ( + BaseRoute, + BaseViewMiddleware, + Connection, + HTTPServer, +) +from traffic_stats import WebTrafficStats +from views import ( + NetworkGetRoute, + NetworkPostRoute, + ReachabilityRoute, + StatusRoute, + TimingMiddleware, +) + + +WEB_TRAFFIC_STATS: WebTrafficStats | None = None + + +def log_ignored_exception(exc: BaseException) -> None: + print(f"pid={os.getpid()} ignoring {exc.__class__.__qualname__}: {exc}") + + +def print_worker_stats() -> None: + stats = WEB_TRAFFIC_STATS + if stats is None: + return + snapshot = json.dumps(stats.snapshot(), sort_keys=True) + print(f"pid={os.getpid()} stats: {snapshot}", flush=True) + + +def force_compile_networkbench_helpers() -> None: + assert HAS_CINDERX + if not cinderx.jit.is_enabled() or cinderx.jit.get_jit_list(): + return + + for func in ( + TimingMiddleware.after_response.__func__, + BaseRoute.prepare.__func__, + BaseViewMiddleware.after_response.__func__, + StatusRoute.handle.__func__, + NetworkPostRoute.prepare.__func__, + NetworkPostRoute.handle.__func__, + NetworkGetRoute.prepare.__func__, + NetworkGetRoute.handle.__func__, + Connection.send_error, + WebTrafficStats.record_request, + WebTrafficStats.snapshot, + WebTrafficStats.status_codes_snapshot, + network_lib.are_reachable, + network_lib.produce_path, + ): + cinderx.jit.force_compile(func) + + +def install_worker_shutdown_handlers() -> None: + def shutdown_signal(_signum, _frame): + try: + loop = asyncio.get_event_loop() + except RuntimeError as exc: + log_ignored_exception(exc) + return + loop.stop() + raise SystemExit(0) + + signal.signal(signal.SIGTERM, shutdown_signal) + signal.signal(signal.SIGINT, shutdown_signal) + + +def create_server_socket() -> socket.socket: + server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + server_socket.bind((config.HOST, config.PORT)) + server_socket.listen() + return server_socket + + +def serve_worker(server_socket: socket.socket) -> None: + global WEB_TRAFFIC_STATS + WEB_TRAFFIC_STATS = WebTrafficStats() + server = HTTPServer( + server_socket, + (StatusRoute, NetworkPostRoute, NetworkGetRoute, ReachabilityRoute), + WEB_TRAFFIC_STATS, + ) + install_worker_shutdown_handlers() + try: + asyncio.run(server.serve_forever()) + except RuntimeError as exc: + log_ignored_exception(exc) + finally: + print_worker_stats() + server_socket.close() + + +def start_server_processes( + server_socket: socket.socket, +) -> list[multiprocessing.Process]: + processes = [ + multiprocessing.Process(target=serve_worker, args=(server_socket,)) + for _ in range(config.SERVER_PROCESS_COUNT) + ] + for process in processes: + process.start() + return processes + + +def stop_server_processes(processes: Sequence[multiprocessing.Process]) -> None: + for process in processes: + if process.is_alive(): + process.terminate() + for process in processes: + process.join() + + +def install_parent_shutdown_handlers( + processes: Sequence[multiprocessing.Process], +) -> None: + def shutdown_signal(_signum, _frame): + stop_server_processes(processes) + print("Bye!") + sys.exit(0) + + signal.signal(signal.SIGTERM, shutdown_signal) + signal.signal(signal.SIGINT, shutdown_signal) + + +def wait_for_server_processes(processes: Sequence[multiprocessing.Process]) -> None: + while any(process.is_alive() for process in processes): + time.sleep(0.5) + + +def main() -> None: + if HAS_CINDERX: + if cinderx.jit.get_jit_list(): + cinderx.jit.precompile_all() + cinderx.jit.disable() + try: + cinderx.enable_parallel_gc() + except RuntimeError: + print("Could not enable parallel gc") + pass + print(f"{cinderx.get_parallel_gc_settings()=}") + else: + cinderx.jit.auto() + force_compile_networkbench_helpers() + multiprocessing.set_start_method("fork") + server_socket = create_server_socket() + processes = start_server_processes(server_socket) + server_socket.close() + install_parent_shutdown_handlers(processes) + try: + wait_for_server_processes(processes) + finally: + stop_server_processes(processes) + print("Bye!") + + +if __name__ == "__main__": + main() diff --git a/cinderx/benchmarks/networkbench/simple_web_framework.py b/cinderx/benchmarks/networkbench/simple_web_framework.py new file mode 100644 index 000000000..c7dd0f391 --- /dev/null +++ b/cinderx/benchmarks/networkbench/simple_web_framework.py @@ -0,0 +1,14 @@ +try: + import cinderx.jit + HAS_CINDERX = cinderx.jit.is_enabled() +except ImportError: + HAS_CINDERX = False + +if HAS_CINDERX: + print("Using simple_web_framework_static") + from cinderx.compiler.strict import loader as static_python_loader + static_python_loader.install() + from simple_web_framework_static import * +else: + print("Using simple_web_framework_py") + from simple_web_framework_py import * diff --git a/cinderx/benchmarks/networkbench/simple_web_framework_py.py b/cinderx/benchmarks/networkbench/simple_web_framework_py.py new file mode 100644 index 000000000..51e58ef17 --- /dev/null +++ b/cinderx/benchmarks/networkbench/simple_web_framework_py.py @@ -0,0 +1,413 @@ +from __future__ import annotations + +import asyncio +from http import HTTPStatus +import json +import os +import socket +import sys +import time +import traceback +from collections.abc import Callable, Mapping +from typing import Any, ClassVar, TypeVar + + +_F = TypeVar("_F", bound=Callable[..., Any]) +HEADER_ENCODING = "iso-8859-1" + + +class Request: + __weakref__: Any + + def __init__( + self, + method: str, + path: str, + version: str, + headers: dict[str, str], + raw_request_line: str, + header_size: int, + body_size: int, + ) -> None: + self.method = method + self.path = path + self.version = version + self.headers = headers + self.raw_request_line = raw_request_line + self.header_size = header_size + self.body_size = body_size + + +class HeaderNames: + BENCHMARK = "X-Networkbench" + CACHE_CONTROL = "Cache-Control" + CONTENT_LENGTH = "Content-Length" + CONTENT_TYPE = "Content-Type" + CONNECTION = "Connection" + PAYLOAD_KEYS = "X-Payload-Keys" + REQUEST_BYTES = "X-Request-Bytes" + RESPONSE_KEYS = "X-Response-Keys" + ROUTE = "X-Route" + VIEW_STACK = "X-View-Stack" + REQUEST_CONTENT_LENGTH = "content-length" + REQUEST_CONTENT_TYPE = "content-type" + + +class RequestContext: + def __init__( + self, + connection: Connection, + route_type: type[BaseRoute], + traffic_stats: Any | None = None, + ) -> None: + self.connection = connection + self.route_type = route_type + self.traffic_stats = traffic_stats + self.payload: object = route_type.no_payload + self.response: object | None = None + self.response_headers: dict[str, str] = {} + self.metadata: dict[str, object] = {} + self.started_ns: int = 0 + + +class BaseViewMiddleware: + enabled: ClassVar[bool] = True + metadata_key: ClassVar[str] = "" + response_header: ClassVar[str] = "" + + @classmethod + async def before_request(cls, context: RequestContext) -> bool: + return cls.enabled + + @classmethod + async def after_response(cls, context: RequestContext) -> object | None: + return context.response + + +class ViewStack: + @classmethod + async def before_view(cls, context: RequestContext) -> bool: + for middleware_type in context.route_type.view_stack: + if not await middleware_type.before_request(context): + return False + return True + + @classmethod + async def after_view(cls, context: RequestContext) -> object | None: + for middleware_type in reversed(context.route_type.view_stack): + response = await middleware_type.after_response(context) + if response is None: + return None + context.response = response + return context.response + + +class BaseRoute: + method: ClassVar[str] = "GET" + route_name: ClassVar[str] = "base" + view_stack_name: ClassVar[str] = "base" + benchmark_name: ClassVar[str] = "networkbench" + cache_control: ClassVar[str] = "no-store" + extra_response_headers: ClassVar[tuple[tuple[str, str], ...]] = () + view_stack: ClassVar[tuple[type[BaseViewMiddleware], ...]] = () + status_code: ClassVar[int] = 200 + response_content_type: ClassVar[str] = "application/json" + error_content_type: ClassVar[str] = "text/plain" + no_payload: ClassVar[object] = object() + + @classmethod + async def prepare(cls, connection: Connection) -> object: + return cls.no_payload + + @classmethod + async def handle(cls, connection: Connection, payload: object) -> object: + return {} + + @classmethod + async def finalize_response(cls, context: RequestContext) -> None: + encoded = await cls.encode_response(context.connection, context.response) + headers = await cls.response_headers(context.connection, encoded, context) + context.connection.send_response(cls.status_code, encoded, headers) + + @classmethod + async def encode_response(cls, connection: Connection, response: object) -> bytes: + return json.dumps(response).encode() + + @classmethod + async def response_headers( + cls, + connection: Connection, + encoded: bytes, + context: RequestContext, + ) -> dict[str, str]: + headers = { + HeaderNames.BENCHMARK: cls.benchmark_name, + HeaderNames.CACHE_CONTROL: cls.cache_control, + HeaderNames.CONTENT_TYPE: cls.response_content_type, + HeaderNames.ROUTE: cls.route_name, + } + headers.update(context.response_headers) + for name, value in cls.extra_response_headers: + headers[name] = value + return headers + + +class Connection(asyncio.Protocol): + NEED_MORE_DATA = object() + + def __init__( + self, + routes: tuple[type[BaseRoute], ...] = (), + traffic_stats: Any | None = None, + ) -> None: + self.routes: tuple[type[BaseRoute], ...] = routes + self.transport: asyncio.Transport | None = None + self.peername: Any | None = None + self.buffer: bytearray = bytearray() + self.request: Request | None = None + self.handle_task: asyncio.Task[None] | None = None + self.response_sent: bool = False + self.traffic_stats = traffic_stats + + def connection_made(self, transport: asyncio.Transport) -> None: + self.transport = transport + self.peername = transport.get_extra_info("peername") + + def data_received(self, data: bytes) -> None: + if self.response_sent: + return + self.buffer.extend(data) + handle_task = self.handle_task + if handle_task is None or handle_task.done(): + self.handle_task = asyncio.create_task(self.handle()) + + async def handle(self) -> None: + try: + if self.request is None: + request, error = await self.read_request() + self.request = request + if self.request is None: + if error is not None: + self.send_error(400, error) + return + + await self.dispatch_request() + except Exception: + traceback.print_exc() + if not self.response_sent: + self.send_error(500) + + async def dispatch_request(self) -> None: + route_type = self.find_route() + if route_type is None: + self.send_error(404) + return + + payload = await route_type.prepare(self) + if payload is self.NEED_MORE_DATA: + return + if payload is None: + return + + context = RequestContext(self, route_type, self.traffic_stats) + context.payload = payload + if not await ViewStack.before_view(context): + return + + response = await route_type.handle(self, payload) + if response is None: + return + context.response = response + response = await ViewStack.after_view(context) + if response is None: + return + await route_type.finalize_response(context) + + def find_route(self) -> type[BaseRoute] | None: + request = self.request + if request is None: + return None + for route_type in self.routes: + if request.method == route_type.method and request.path == route_type.path: + return route_type + return None + + async def read_request(self) -> tuple[Request | None, str | None]: + header_end = self.buffer.find(b"\r\n\r\n") + separator_length = 4 + if header_end < 0: + header_end = self.buffer.find(b"\n\n") + separator_length = 2 + if header_end < 0: + return None, None + + request_header_size = header_end + separator_length + encoded_headers = bytes(self.buffer[:header_end]) + del self.buffer[: header_end + separator_length] + try: + decoded_headers = encoded_headers.decode(HEADER_ENCODING) + except UnicodeDecodeError as exc: + return None, str(exc) + + lines = decoded_headers.splitlines() + if not lines: + return None, "Bad request syntax" + + decoded_request_line = lines[0] + parts = decoded_request_line.split() + if len(parts) != 3: + return None, "Bad request syntax" + + headers: dict[str, str] = {} + for line in lines[1:]: + if not line: + continue + if ":" not in line: + return None, "Bad header syntax" + name, value = line.split(":", 1) + headers[name.strip().lower()] = value.strip() + + request_body_size = 0 + content_length = headers.get(HeaderNames.REQUEST_CONTENT_LENGTH) + if content_length is not None: + try: + request_body_size = int(content_length) + except ValueError: + request_body_size = 0 + if request_body_size < 0: + request_body_size = 0 + + return ( + Request( + parts[0], + parts[1], + parts[2], + headers, + decoded_request_line, + request_header_size, + request_body_size, + ), + None, + ) + + def send_response( + self, + status_code: int | HTTPStatus, + body: bytes = b"", + headers: Mapping[str, str] | None = None, + ) -> None: + status = HTTPStatus(status_code) + response_headers = { + HeaderNames.CONTENT_LENGTH: str(len(body)), + HeaderNames.CONNECTION: "close", + } + if headers is not None: + response_headers.update(headers) + + header_chunks = [ + f"HTTP/1.1 {status.value} {status.phrase}\r\n".encode("ascii") + ] + for name, value in response_headers.items(): + header_chunks.append(f"{name}: {value}\r\n".encode("ascii")) + header_chunks.append(b"\r\n") + + encoded_headers = b"".join(header_chunks) + transport = self.transport + assert transport is not None + transport.write(encoded_headers) + transport.write(body) + stats = self.traffic_stats + if stats is not None: + request_header_size = 0 + request_body_size = 0 + if self.request is not None: + request: Request = self.request + request_header_size = request.header_size + request_body_size = request.body_size + stats.record_request(request_header_size, request_body_size) + stats.record_response(status.value, len(encoded_headers), len(body)) + self.log_request(status.value, len(body)) + self.response_sent = True + transport.close() + + def send_error( + self, + status_code: int, + message: str | None = None, + route_type: type[BaseRoute] = BaseRoute, + ) -> None: + status = HTTPStatus(status_code) + _message: str = "" + if message is None: + _message = status.phrase + else: + _message = message + self.send_response( + status, + _message.encode(), + {HeaderNames.CONTENT_TYPE: route_type.error_content_type}, + ) + + def send_json_response( + self, + response: object, + route_type: type[BaseRoute] = BaseRoute, + ) -> None: + encoded = json.dumps(response).encode() + self.send_response( + route_type.status_code, + encoded, + {HeaderNames.CONTENT_TYPE: route_type.response_content_type}, + ) + + def address_string(self) -> str: + if not self.peername: + return "-" + return self.peername[0] + + def log_date_time_string(self) -> str: + return time.strftime("%d/%b/%Y %H:%M:%S", time.localtime()) + + def log_request(self, status_code: int, response_size: int | str = "-") -> None: + if self.request is None: + request_line = "" + else: + request_line = self.request.raw_request_line + sys.stderr.write( + "%s - pid=%s - [%s] \"%s\" %s %s\n" + % ( + self.address_string(), + os.getpid(), + self.log_date_time_string(), + request_line, + status_code, + response_size, + ) + ) + + +class HTTPServer: + def __init__( + self, + server_socket: socket.socket, + routes: tuple[type[BaseRoute], ...] = (), + traffic_stats: Any | None = None, + ) -> None: + self.server_socket = server_socket + self.routes = routes + self.traffic_stats = traffic_stats + self.server: asyncio.Server | None = None + + def create_connection(self) -> Connection: + return Connection(self.routes, self.traffic_stats) + + async def serve_forever(self) -> None: + self.server_socket.setblocking(False) + loop = asyncio.get_running_loop() + self.server = await loop.create_server( + self.create_connection, + sock=self.server_socket, + ) + print(f"pid={os.getpid()} Server listening at {self.server.sockets[0].getsockname()}") + async with self.server: + await self.server.serve_forever() diff --git a/cinderx/benchmarks/networkbench/simple_web_framework_static.py b/cinderx/benchmarks/networkbench/simple_web_framework_static.py new file mode 100644 index 000000000..c79ebecd8 --- /dev/null +++ b/cinderx/benchmarks/networkbench/simple_web_framework_static.py @@ -0,0 +1,420 @@ +from __future__ import annotations + +import __static__ + +import asyncio +from http import HTTPStatus +import json +import os +import socket +import sys +import time +import traceback +from collections.abc import Callable, Mapping +from typing import Any, ClassVar, TypeVar +import cinderx.jit + + +_F = TypeVar("_F", bound=Callable[..., Any]) + +HEADER_ENCODING = "iso-8859-1" + + +class Request: + __weakref__: Any + + def __init__( + self, + method: str, + path: str, + version: str, + headers: dict[str, str], + raw_request_line: str, + header_size: int, + body_size: int, + ) -> None: + self.method = method + self.path = path + self.version = version + self.headers = headers + self.raw_request_line = raw_request_line + self.header_size = header_size + self.body_size = body_size + + +class HeaderNames: + BENCHMARK = "X-Networkbench" + CACHE_CONTROL = "Cache-Control" + CONTENT_LENGTH = "Content-Length" + CONTENT_TYPE = "Content-Type" + CONNECTION = "Connection" + PAYLOAD_KEYS = "X-Payload-Keys" + REQUEST_BYTES = "X-Request-Bytes" + RESPONSE_KEYS = "X-Response-Keys" + ROUTE = "X-Route" + VIEW_STACK = "X-View-Stack" + REQUEST_CONTENT_LENGTH = "content-length" + REQUEST_CONTENT_TYPE = "content-type" + + +class RequestContext: + def __init__( + self, + connection: Connection, + route_type: type[BaseRoute], + traffic_stats: Any | None = None, + ) -> None: + self.connection = connection + self.route_type = route_type + self.traffic_stats = traffic_stats + self.payload: object = route_type.no_payload + self.response: object | None = None + self.response_headers: dict[str, str] = {} + self.metadata: dict[str, object] = {} + self.started_ns: int = 0 + + +class BaseViewMiddleware: + enabled: ClassVar[bool] = True + metadata_key: ClassVar[str] = "" + response_header: ClassVar[str] = "" + + @classmethod + async def before_request(cls, context: RequestContext) -> bool: + return cls.enabled + + @classmethod + async def after_response(cls, context: RequestContext) -> object | None: + return context.response + + +class ViewStack: + @classmethod + async def before_view(cls, context: RequestContext) -> bool: + for middleware_type in context.route_type.view_stack: + if not await middleware_type.before_request(context): + return False + return True + + @classmethod + async def after_view(cls, context: RequestContext) -> object | None: + for middleware_type in reversed(context.route_type.view_stack): + response = await middleware_type.after_response(context) + if response is None: + return None + context.response = response + return context.response + + +class BaseRoute: + method: ClassVar[str] = "GET" + route_name: ClassVar[str] = "base" + view_stack_name: ClassVar[str] = "base" + benchmark_name: ClassVar[str] = "networkbench" + cache_control: ClassVar[str] = "no-store" + extra_response_headers: ClassVar[tuple[tuple[str, str], ...]] = () + view_stack: ClassVar[tuple[type[BaseViewMiddleware], ...]] = () + status_code: ClassVar[int] = 200 + response_content_type: ClassVar[str] = "application/json" + error_content_type: ClassVar[str] = "text/plain" + no_payload: ClassVar[object] = object() + + @classmethod + async def prepare(cls, connection: Connection) -> object: + return cls.no_payload + + @classmethod + async def handle(cls, connection: Connection, payload: object) -> object: + return {} + + @classmethod + async def finalize_response(cls, context: RequestContext) -> None: + encoded = await cls.encode_response(context.connection, context.response) + headers = await cls.response_headers(context.connection, encoded, context) + context.connection.send_response(cls.status_code, encoded, headers) + + @classmethod + async def encode_response(cls, connection: Connection, response: object) -> bytes: + return json.dumps(response).encode() + + @classmethod + async def response_headers( + cls, + connection: Connection, + encoded: bytes, + context: RequestContext, + ) -> dict[str, str]: + headers = { + HeaderNames.BENCHMARK: cls.benchmark_name, + HeaderNames.CACHE_CONTROL: cls.cache_control, + HeaderNames.CONTENT_TYPE: cls.response_content_type, + HeaderNames.ROUTE: cls.route_name, + } + headers.update(context.response_headers) + for name, value in cls.extra_response_headers: + headers[name] = value + return headers + + +class Connection(asyncio.Protocol): + NEED_MORE_DATA = object() + + def __init__( + self, + routes: tuple[type[BaseRoute], ...] = (), + traffic_stats: Any | None = None, + ) -> None: + self.routes: tuple[type[BaseRoute], ...] = routes + self.transport: asyncio.Transport | None = None + self.peername: Any | None = None + self.buffer: bytearray = bytearray() + self.request: Request | None = None + self.handle_task: asyncio.Task[None] | None = None + self.response_sent: bool = False + self.traffic_stats = traffic_stats + + def connection_made(self, transport: asyncio.Transport) -> None: + self.transport = transport + self.peername = transport.get_extra_info("peername") + + def data_received(self, data: bytes) -> None: + if self.response_sent: + return + self.buffer.extend(data) + handle_task = self.handle_task + if handle_task is None or handle_task.done(): + self.handle_task = asyncio.create_task(self.handle()) + + async def handle(self) -> None: + try: + if self.request is None: + request, error = await self.read_request() + self.request = request + if self.request is None: + if error is not None: + self.send_error(400, error) + return + + await self.dispatch_request() + except Exception: + traceback.print_exc() + if not self.response_sent: + self.send_error(500) + + async def dispatch_request(self) -> None: + route_type = self.find_route() + if route_type is None: + self.send_error(404) + return + + payload = await route_type.prepare(self) + if payload is self.NEED_MORE_DATA: + return + if payload is None: + return + + context = RequestContext(self, route_type, self.traffic_stats) + context.payload = payload + if not await ViewStack.before_view(context): + return + + response = await route_type.handle(self, payload) + if response is None: + return + context.response = response + response = await ViewStack.after_view(context) + if response is None: + return + await route_type.finalize_response(context) + + def find_route(self) -> type[BaseRoute] | None: + request = self.request + if request is None: + return None + for route_type in self.routes: + if request.method == route_type.method and request.path == route_type.path: + return route_type + return None + + async def read_request(self) -> tuple[Request | None, str | None]: + header_end = self.buffer.find(b"\r\n\r\n") + separator_length = 4 + if header_end < 0: + header_end = self.buffer.find(b"\n\n") + separator_length = 2 + if header_end < 0: + return None, None + + request_header_size = header_end + separator_length + encoded_headers = bytes(self.buffer[:header_end]) + del self.buffer[: header_end + separator_length] + try: + decoded_headers = encoded_headers.decode(HEADER_ENCODING) + except UnicodeDecodeError as exc: + return None, str(exc) + + lines = decoded_headers.splitlines() + if not lines: + return None, "Bad request syntax" + + decoded_request_line = lines[0] + parts = decoded_request_line.split() + if len(parts) != 3: + return None, "Bad request syntax" + + headers: dict[str, str] = {} + for line in lines[1:]: + if not line: + continue + if ":" not in line: + return None, "Bad header syntax" + name, value = line.split(":", 1) + headers[name.strip().lower()] = value.strip() + + request_body_size = 0 + content_length = headers.get(HeaderNames.REQUEST_CONTENT_LENGTH) + if content_length is not None: + try: + request_body_size = int(content_length) + except ValueError: + request_body_size = 0 + if request_body_size < 0: + request_body_size = 0 + + return ( + Request( + parts[0], + parts[1], + parts[2], + headers, + decoded_request_line, + request_header_size, + request_body_size, + ), + None, + ) + + def send_response( + self, + status_code: int | HTTPStatus, + body: bytes = b"", + headers: Mapping[str, str] | None = None, + ) -> None: + status = HTTPStatus(status_code) + response_headers = { + HeaderNames.CONTENT_LENGTH: str(len(body)), + HeaderNames.CONNECTION: "close", + } + if headers is not None: + response_headers.update(headers) + + header_chunks = [ + f"HTTP/1.1 {status.value} {status.phrase}\r\n".encode("ascii") + ] + for name, value in response_headers.items(): + header_chunks.append(f"{name}: {value}\r\n".encode("ascii")) + header_chunks.append(b"\r\n") + + encoded_headers = b"".join(header_chunks) + transport = self.transport + assert transport is not None + transport.write(encoded_headers) + transport.write(body) + stats = self.traffic_stats + if stats is not None: + request_header_size = 0 + request_body_size = 0 + if self.request is not None: + request: Request = self.request + request_header_size = request.header_size + request_body_size = request.body_size + stats.record_request(request_header_size, request_body_size) + stats.record_response(status.value, len(encoded_headers), len(body)) + self.log_request(status.value, len(body)) + self.response_sent = True + transport.close() + + def send_error( + self, + status_code: int, + message: str | None = None, + route_type: type[BaseRoute] = BaseRoute, + ) -> None: + status = HTTPStatus(status_code) + _message: str = "" + if message is None: + _message = status.phrase + else: + _message = message + self.send_response( + status, + _message.encode(), + {HeaderNames.CONTENT_TYPE: route_type.error_content_type}, + ) + + def send_json_response( + self, + response: object, + route_type: type[BaseRoute] = BaseRoute, + ) -> None: + encoded = json.dumps(response).encode() + self.send_response( + route_type.status_code, + encoded, + {HeaderNames.CONTENT_TYPE: route_type.response_content_type}, + ) + + @cinderx.jit.jit_suppress + def address_string(self) -> str: + if not self.peername: + return "-" + return self.peername[0] + + @cinderx.jit.jit_suppress + def log_date_time_string(self) -> str: + return time.strftime("%d/%b/%Y %H:%M:%S", time.localtime()) + + @cinderx.jit.jit_suppress + def log_request(self, status_code: int, response_size: int | str = "-") -> None: + if self.request is None: + request_line = "" + else: + request_line = self.request.raw_request_line + sys.stderr.write( + "%s - pid=%s - [%s] \"%s\" %s %s\n" + % ( + self.address_string(), + os.getpid(), + self.log_date_time_string(), + request_line, + status_code, + response_size, + ) + ) + + +class HTTPServer: + def __init__( + self, + server_socket: socket.socket, + routes: tuple[type[BaseRoute], ...] = (), + traffic_stats: Any | None = None, + ) -> None: + self.server_socket = server_socket + self.routes = routes + self.traffic_stats = traffic_stats + self.server: asyncio.Server | None = None + + def create_connection(self) -> Connection: + return Connection(self.routes, self.traffic_stats) + + async def serve_forever(self) -> None: + self.server_socket.setblocking(False) + loop = asyncio.get_running_loop() + self.server = await loop.create_server( + self.create_connection, + sock=self.server_socket, + ) + print(f"pid={os.getpid()} Server listening at {self.server.sockets[0].getsockname()}") + async with self.server: + await self.server.serve_forever() diff --git a/cinderx/benchmarks/networkbench/traffic_stats.py b/cinderx/benchmarks/networkbench/traffic_stats.py new file mode 100644 index 000000000..1478dad8a --- /dev/null +++ b/cinderx/benchmarks/networkbench/traffic_stats.py @@ -0,0 +1,14 @@ +try: + import cinderx.jit + HAS_CINDERX = cinderx.jit.is_enabled() +except ImportError: + HAS_CINDERX = False + +if HAS_CINDERX: + print("Using traffic_stats_static") + from cinderx.compiler.strict import loader as static_python_loader + static_python_loader.install() + from traffic_stats_static import * +else: + print("Using traffic_stats_py") + from traffic_stats_py import * diff --git a/cinderx/benchmarks/networkbench/traffic_stats_py.py b/cinderx/benchmarks/networkbench/traffic_stats_py.py new file mode 100644 index 000000000..c232a8458 --- /dev/null +++ b/cinderx/benchmarks/networkbench/traffic_stats_py.py @@ -0,0 +1,90 @@ +MAX_STATUS_CODE = 600 + + +class WebTrafficStats: + def __init__(self) -> None: + self.request_count = 0 + self.informational_count = 0 + self.success_count = 0 + self.redirect_count = 0 + self.client_error_count = 0 + self.server_error_count = 0 + self.other_status_count = 0 + self.total_request_header_bytes = 0 + self.total_request_body_bytes = 0 + self.total_header_bytes = 0 + self.total_body_bytes = 0 + self.total_response_bytes = 0 + self.timed_request_count = 0 + self.total_request_ns = 0 + self.max_request_ns = 0 + self.status_code_counts = [0] * MAX_STATUS_CODE + + def record_request(self, header_size: int, body_size: int) -> None: + self.request_count += 1 + self.total_request_header_bytes += header_size + self.total_request_body_bytes += body_size + + def record_response( + self, status_code: int, header_size: int, body_size: int + ) -> None: + self.total_header_bytes += header_size + self.total_body_bytes += body_size + self.total_response_bytes += header_size + body_size + + if 0 <= status_code < MAX_STATUS_CODE: + self.status_code_counts[status_code] += 1 + + if 100 <= status_code < 200: + self.informational_count += 1 + elif 200 <= status_code < 300: + self.success_count += 1 + elif 300 <= status_code < 400: + self.redirect_count += 1 + elif 400 <= status_code < 500: + self.client_error_count += 1 + elif 500 <= status_code < 600: + self.server_error_count += 1 + else: + self.other_status_count += 1 + + def record_request_elapsed(self, elapsed_ns: int) -> None: + self.timed_request_count += 1 + self.total_request_ns += elapsed_ns + if elapsed_ns > self.max_request_ns: + self.max_request_ns = elapsed_ns + + def snapshot(self) -> dict[str, object]: + return { + "requests": { + "count": self.request_count, + "header_bytes": self.total_request_header_bytes, + "body_bytes": self.total_request_body_bytes, + }, + "responses": { + "status_classes": { + "1xx": self.informational_count, + "2xx": self.success_count, + "3xx": self.redirect_count, + "4xx": self.client_error_count, + "5xx": self.server_error_count, + "other": self.other_status_count, + }, + "status_codes": self.status_codes_snapshot(), + "header_bytes": self.total_header_bytes, + "body_bytes": self.total_body_bytes, + "total_bytes": self.total_response_bytes, + }, + "request_timing": { + "count": self.timed_request_count, + "total_ns": self.total_request_ns, + "max_ns": self.max_request_ns, + }, + } + + def status_codes_snapshot(self) -> dict[str, int]: + return { + str(status_code): count + for status_code, count in enumerate(self.status_code_counts) + if count + } diff --git a/cinderx/benchmarks/networkbench/traffic_stats_static.py b/cinderx/benchmarks/networkbench/traffic_stats_static.py new file mode 100644 index 000000000..1827efacc --- /dev/null +++ b/cinderx/benchmarks/networkbench/traffic_stats_static.py @@ -0,0 +1,128 @@ +import __static__ +from __static__ import Array, box, int64 + + +MAX_STATUS_CODE = 600 + + +class WebTrafficStats: + request_count: int64 + informational_count: int64 + success_count: int64 + redirect_count: int64 + client_error_count: int64 + server_error_count: int64 + other_status_count: int64 + total_request_header_bytes: int64 + total_request_body_bytes: int64 + total_header_bytes: int64 + total_body_bytes: int64 + total_response_bytes: int64 + timed_request_count: int64 + total_request_ns: int64 + max_request_ns: int64 + status_code_counts: Array[int64] + + def __init__(self) -> None: + self.request_count = 0 + self.informational_count = 0 + self.success_count = 0 + self.redirect_count = 0 + self.client_error_count = 0 + self.server_error_count = 0 + self.other_status_count = 0 + self.total_request_header_bytes = 0 + self.total_request_body_bytes = 0 + self.total_header_bytes = 0 + self.total_body_bytes = 0 + self.total_response_bytes = 0 + self.timed_request_count = 0 + self.total_request_ns = 0 + self.max_request_ns = 0 + self.status_code_counts = Array[int64](MAX_STATUS_CODE) + + def record_request(self, header_size: int, body_size: int) -> None: + headers: int64 = int64(header_size) + body: int64 = int64(body_size) + + self.request_count += 1 + self.total_request_header_bytes += headers + self.total_request_body_bytes += body + + def record_response( + self, status_code: int, header_size: int, body_size: int + ) -> None: + code: int64 = int64(status_code) + headers: int64 = int64(header_size) + body: int64 = int64(body_size) + + self.total_header_bytes += headers + self.total_body_bytes += body + self.total_response_bytes += headers + body + + if code >= 0: + if code < int64(MAX_STATUS_CODE): + self.status_code_counts[code] += 1 + + if code >= 100: + if code < 200: + self.informational_count += 1 + return + if code < 300: + self.success_count += 1 + return + if code < 400: + self.redirect_count += 1 + return + if code < 500: + self.client_error_count += 1 + return + if code < 600: + self.server_error_count += 1 + return + self.other_status_count += 1 + + def record_request_elapsed(self, elapsed_ns: int) -> None: + elapsed: int64 = int64(elapsed_ns) + self.timed_request_count += 1 + self.total_request_ns += elapsed + if elapsed > self.max_request_ns: + self.max_request_ns = elapsed + + def snapshot(self) -> dict[str, object]: + return { + "requests": { + "count": box(self.request_count), + "header_bytes": box(self.total_request_header_bytes), + "body_bytes": box(self.total_request_body_bytes), + }, + "responses": { + "status_classes": { + "1xx": box(self.informational_count), + "2xx": box(self.success_count), + "3xx": box(self.redirect_count), + "4xx": box(self.client_error_count), + "5xx": box(self.server_error_count), + "other": box(self.other_status_count), + }, + "status_codes": self.status_codes_snapshot(), + "header_bytes": box(self.total_header_bytes), + "body_bytes": box(self.total_body_bytes), + "total_bytes": box(self.total_response_bytes), + }, + "request_timing": { + "count": box(self.timed_request_count), + "total_ns": box(self.total_request_ns), + "max_ns": box(self.max_request_ns), + }, + } + + def status_codes_snapshot(self) -> dict[str, int]: + status_codes = {} + code: int64 = 0 + while code < int64(MAX_STATUS_CODE): + count: int64 = self.status_code_counts[code] + if count: + status_codes[str(box(code))] = box(count) + code += 1 + return status_codes diff --git a/cinderx/benchmarks/networkbench/very_simple_queue.py b/cinderx/benchmarks/networkbench/very_simple_queue.py new file mode 100644 index 000000000..c012f585d --- /dev/null +++ b/cinderx/benchmarks/networkbench/very_simple_queue.py @@ -0,0 +1,34 @@ +from __future__ import annotations + + +class Item: + def __init__(self, v: int, next: Item | None = None) -> None: + self.v = v + self.next = next + + +class VerySimpleQueue: + def __init__(self) -> None: + self.head: Item | None = None + self.tail: Item | None = None + + def put(self, v: int) -> None: + item = Item(v) + if self.tail is None: + self.head = item + self.tail = item + return + self.tail.next = item + self.tail = item + + def get(self) -> int: + item = self.head + if item is None: + raise IndexError("get from empty queue") + self.head = item.next + if self.head is None: + self.tail = None + return item.v + + def empty(self) -> bool: + return self.head is None diff --git a/cinderx/benchmarks/networkbench/views.py b/cinderx/benchmarks/networkbench/views.py new file mode 100644 index 000000000..4ce190fda --- /dev/null +++ b/cinderx/benchmarks/networkbench/views.py @@ -0,0 +1,444 @@ +import asyncio +import os +import time +import traceback +from typing import TypedDict + +import config +import matrix_codec +import network_lib +from simple_web_framework import ( + BaseRoute, + BaseViewMiddleware, + Connection, + HeaderNames, + RequestContext, +) + + +class NetworkPostPayload(TypedDict): + network_id: int + body: bytes + + +class RouteMetadataMiddleware(BaseViewMiddleware): + metadata_key = "route" + + @classmethod + async def before_request(cls, context: RequestContext) -> bool: + route_type = context.route_type + context.metadata[cls.metadata_key] = route_type.route_name + return cls.enabled + + @classmethod + async def after_response(cls, context: RequestContext) -> object | None: + route_type = context.route_type + context.response_headers.update( + { + HeaderNames.ROUTE: str( + context.metadata.get( + cls.metadata_key, + route_type.route_name, + ) + ), + HeaderNames.VIEW_STACK: route_type.view_stack_name, + } + ) + return context.response + + +class RequestHeadersMiddleware(BaseViewMiddleware): + metadata_key = "request_bytes" + default_length = "0" + + @classmethod + async def before_request(cls, context: RequestContext) -> bool: + request = context.connection.request + if request is None: + context.metadata[cls.metadata_key] = cls.default_length + return cls.enabled + + value = request.headers.get(HeaderNames.REQUEST_CONTENT_LENGTH) + if value is None: + value = cls.default_length + context.metadata[cls.metadata_key] = value + return cls.enabled + + @classmethod + async def after_response(cls, context: RequestContext) -> object | None: + context.response_headers.update( + { + HeaderNames.REQUEST_BYTES: str( + context.metadata.get( + cls.metadata_key, + cls.default_length, + ) + ) + } + ) + return context.response + + +class PayloadAuditMiddleware(BaseViewMiddleware): + empty_size = "0" + + @classmethod + async def after_response(cls, context: RequestContext) -> object | None: + payload = context.payload + if isinstance(payload, dict): + payload_keys = str(len(payload)) + else: + payload_keys = cls.empty_size + context.response_headers.update({HeaderNames.PAYLOAD_KEYS: payload_keys}) + return context.response + + +class ResponseHeadersMiddleware(BaseViewMiddleware): + empty_size = "0" + + @classmethod + async def after_response(cls, context: RequestContext) -> object | None: + route_type = context.route_type + response = context.response + if isinstance(response, dict): + response_keys = str(len(response)) + else: + response_keys = cls.empty_size + context.response_headers.update( + { + HeaderNames.BENCHMARK: route_type.benchmark_name, + HeaderNames.CACHE_CONTROL: route_type.cache_control, + HeaderNames.CONTENT_TYPE: route_type.response_content_type, + HeaderNames.RESPONSE_KEYS: response_keys, + } + ) + return response + + +class TimingMiddleware(BaseViewMiddleware): + min_elapsed_ns = 0 + + @classmethod + async def before_request(cls, context: RequestContext) -> bool: + if cls.enabled: + context.started_ns = time.perf_counter_ns() + return cls.enabled + + @classmethod + async def after_response(cls, context: RequestContext) -> object | None: + started_ns = context.started_ns + if started_ns: + elapsed_ns = time.perf_counter_ns() - started_ns + if elapsed_ns < cls.min_elapsed_ns: + elapsed_ns = cls.min_elapsed_ns + stats = context.traffic_stats + if stats is not None: + stats.record_request_elapsed(elapsed_ns) + return context.response + + +class NetworkbenchRoute(BaseRoute): + route_name = "base" + view_stack_name = "base" + benchmark_name = "networkbench" + cache_control = "no-store" + extra_response_headers = () + view_stack = ( + TimingMiddleware, + RouteMetadataMiddleware, + RequestHeadersMiddleware, + PayloadAuditMiddleware, + ResponseHeadersMiddleware, + ) + status_code = 200 + response_content_type = config.JSON_CONTENT_TYPE + error_content_type = "text/plain" + + +class StatusRoute(NetworkbenchRoute): + path = config.STATUS_PATH + route_name = "status" + view_stack_name = "status.view" + + @classmethod + async def handle( + cls, + connection: Connection, + payload: object, + ) -> dict[str, object]: + stats = connection.traffic_stats + if stats is None: + return {} + return stats.snapshot() + + +class NetworkRoute(NetworkbenchRoute): + path = config.NETWORK_PATH + route_name = "network" + view_stack_name = "network.view" + request_content_type = config.NETWORK_CONTENT_TYPE + response_content_type = config.NETWORK_CONTENT_TYPE + max_body_bytes = config.MAX_REQUEST_BODY_BYTES + request_network_id_header = "x-network-id" + request_network_size_header = "x-network-size" + response_network_id_header = "X-Network-Id" + + @classmethod + def parse_int_header( + cls, + connection: Connection, + header_name: str, + label: str, + ) -> int | None: + request = connection.request + assert request is not None + value = request.headers.get(header_name) + if value is None: + connection.send_error(400, f"Missing {label}", cls) + return None + try: + parsed = int(value) + except ValueError: + connection.send_error(400, f"Invalid {label}", cls) + return None + if parsed < 0: + connection.send_error(400, f"Invalid {label}", cls) + return None + return parsed + + @classmethod + def parse_network_id(cls, connection: Connection) -> int | None: + return cls.parse_int_header( + connection, + cls.request_network_id_header, + "X-Network-Id", + ) + + @classmethod + def matrix_path(cls, network_id: int) -> str: + return os.path.join(config.NETWORK_STORAGE_DIR, f"network-{network_id}.nbm") + + @classmethod + def write_matrix_file(cls, network_id: int, body: bytes) -> int: + os.makedirs(config.NETWORK_STORAGE_DIR, exist_ok=True) + path = cls.matrix_path(network_id) + tmp_path = f"{path}.{os.getpid()}.tmp" + with open(tmp_path, "wb") as file: + file.write(body) + os.replace(tmp_path, path) + return len(body) + + @classmethod + def read_matrix_file(cls, network_id: int) -> bytes: + with open(cls.matrix_path(network_id), "rb") as file: + return file.read() + + @classmethod + async def write_matrix(cls, network_id: int, body: bytes) -> int: + loop = asyncio.get_running_loop() + return await loop.run_in_executor(None, cls.write_matrix_file, network_id, body) + + @classmethod + async def read_matrix(cls, network_id: int) -> bytes: + loop = asyncio.get_running_loop() + return await loop.run_in_executor(None, cls.read_matrix_file, network_id) + + @classmethod + async def response_headers( + cls, + connection: Connection, + encoded: bytes, + context: RequestContext, + ) -> dict[str, str]: + headers = await super().response_headers(connection, encoded, context) + if isinstance(context.payload, int): + headers[cls.response_network_id_header] = str(context.payload) + return headers + + +class NetworkPostRoute(NetworkRoute): + method = "POST" + route_name = "network-post" + view_stack_name = "network.post.view" + response_content_type = config.JSON_CONTENT_TYPE + + @classmethod + async def prepare(cls, connection: Connection) -> object | None: + network_id = cls.parse_network_id(connection) + if network_id is None: + return None + + request = connection.request + assert request is not None + try: + length = int( + request.headers.get( + HeaderNames.REQUEST_CONTENT_LENGTH, + "0", + ) + ) + except ValueError: + connection.send_error(400, "Invalid Content-Length", cls) + return None + + matrix_size = cls.parse_int_header( + connection, + cls.request_network_size_header, + "X-Network-Size", + ) + if matrix_size is None: + return None + + if length >= cls.max_body_bytes: + connection.send_error(413, route_type=cls) + return None + + if ( + request.headers.get(HeaderNames.REQUEST_CONTENT_TYPE, "") + != cls.request_content_type + ): + connection.send_error(400, "Invalid Content-Type", cls) + return None + + if len(connection.buffer) < length: + return connection.NEED_MORE_DATA + + body = bytes(connection.buffer[:length]) + del connection.buffer[:length] + if len(body) != matrix_size * matrix_size: + connection.send_error(400, "Invalid matrix size", cls) + return None + return {"network_id": network_id, "body": body} + + @classmethod + async def handle( + cls, + connection: Connection, + payload: NetworkPostPayload, + ) -> dict[str, int]: + body_size = await cls.write_matrix(payload["network_id"], payload["body"]) + return {"network_id": payload["network_id"], "bytes": body_size} + + +class NetworkGetRoute(NetworkRoute): + method = "GET" + route_name = "network-get" + view_stack_name = "network.get.view" + + @classmethod + async def prepare(cls, connection: Connection) -> int | None: + return cls.parse_network_id(connection) + + @classmethod + async def handle(cls, connection: Connection, network_id: int) -> bytes | None: + try: + return await cls.read_matrix(network_id) + except FileNotFoundError: + connection.send_error(404, route_type=cls) + return None + + @classmethod + async def encode_response(cls, connection: Connection, response: bytes) -> bytes: + return response + + +class ReachabilityRoute(NetworkbenchRoute): + path = config.REACHABLE_PATH + route_name = "reachability" + view_stack_name = "reachability.view" + request_content_type = config.REACHABILITY_CONTENT_TYPE + max_body_bytes = config.MAX_REQUEST_BODY_BYTES + + @classmethod + async def prepare(cls, connection: Connection) -> object | None: + req = await cls.read_request(connection) + if req is connection.NEED_MORE_DATA: + return req + if req is None: + return None + + valid, reason = cls.validate_request(req) + if not valid: + connection.send_error(400, reason, cls) + return None + return req + + @classmethod + async def read_request(cls, connection: Connection) -> object | None: + request = connection.request + assert request is not None + try: + length = int( + request.headers.get( + HeaderNames.REQUEST_CONTENT_LENGTH, + "0", + ) + ) + except ValueError: + connection.send_error(400, "Invalid Content-Length", cls) + return None + + if length >= cls.max_body_bytes: + connection.send_error(413, route_type=cls) + return None + + if ( + request.headers.get(HeaderNames.REQUEST_CONTENT_TYPE, "") + != cls.request_content_type + ): + connection.send_error(400, "Invalid Content-Type", cls) + return None + + if len(connection.buffer) < length: + return connection.NEED_MORE_DATA + + body = bytes(connection.buffer[:length]) + del connection.buffer[:length] + try: + return matrix_codec.decode_reachability_request(body) + except ValueError as exc: + connection.send_error(400, str(exc), cls) + return None + + @classmethod + def validate_request(cls, req: object) -> tuple[bool, str]: + if not isinstance(req, dict): + return False, "not isinstance(req, dict)" + if "source" not in req: + return False, '"source" not in req' + if not isinstance(req["source"], int): + return False, "not isinstance(source, int)" + if "destination" not in req: + return False, '"destination" not in req' + if not isinstance(req["destination"], int): + return False, "not isinstance(destination, int)" + if "graph" not in req: + return False, '"graph" not in req' + if not isinstance(req["graph"], list): + return False, "not isinstance(graph, list)" + if not req["graph"]: + return False, "graph is empty" + node_count = len(req["graph"]) + if req["source"] < 0 or req["source"] >= node_count: + return False, "source is out of range" + if req["destination"] < 0 or req["destination"] >= node_count: + return False, "destination is out of range" + return True, "" + + @classmethod + async def handle( + cls, + connection: Connection, + req: matrix_codec.ReachabilityRequest, + ) -> dict[str, object] | None: + try: + reachable, path = network_lib.are_reachable( + req["graph"], + req["source"], + req["destination"], + ) + except Exception: + traceback.print_exc() + connection.send_error(500, route_type=cls) + return None + return {"reachable": reachable, "path": path} diff --git a/setup.py b/setup.py index 5094653bd..db59ebe7b 100644 --- a/setup.py +++ b/setup.py @@ -13,6 +13,7 @@ import os.path import platform import re +import shlex import shutil import subprocess import sys @@ -192,11 +193,15 @@ def print_section(title: str) -> None: else: workload_env["PYTHONPATH"] = cinderx_so_dir - # Uses the same default workload as CPython's PGO - workload_cmd = [ - sys.executable, - "-c", - """ + profile_task = os.environ.get("CINDERX_PGO_PROFILE_TASK") + if profile_task: + workload_cmd = [sys.executable, *shlex.split(profile_task)] + else: + # Uses the same default workload as CPython's PGO. + workload_cmd = [ + sys.executable, + "-c", + """ import cinderx import sys @@ -210,8 +215,8 @@ def main(): if __name__ == "__main__": main() - """, - ] + """, + ] print(f"Running workload with PYTHONPATH={workload_env['PYTHONPATH']}") workload_args = {