diff --git a/.github/workflows/benchmark.yml b/.github/workflows/benchmark.yml index 8aebd34..221b36b 100644 --- a/.github/workflows/benchmark.yml +++ b/.github/workflows/benchmark.yml @@ -152,8 +152,77 @@ jobs: path: ${{ steps.names.outputs.result_file }} retention-days: 30 + benchmark-python: + runs-on: ubuntu-latest + + strategy: + fail-fast: false + matrix: + driver: + - configs/drivers/default/redis-py.json + - configs/drivers/default/valkey-glide-python.json + workload: + - configs/workloads/reference/basic-standalone-single-client-1M-reqs.json + + steps: + - uses: actions/checkout@v4 + + - name: Set up Python + uses: actions/setup-python@v5 + with: + python-version: '3.12' + + - name: Install build dependencies + run: | + sudo apt-get update + sudo apt-get install -y build-essential + + - name: Install Python engine + run: cd python && pip install -e . + + - name: Start Valkey server + run: | + # Use Makefile target which builds from source and configures with persistence disabled + make server-standalone-start + # Wait for server to be ready + sleep 2 + # Verify server is up and persistence is disabled + work/valkey/bin/valkey-cli ping + work/valkey/bin/valkey-cli CONFIG GET save + + - name: Extract names for result file + id: names + run: | + DRIVER_NAME=$(basename ${{ matrix.driver }} .json) + WORKLOAD_NAME=$(basename ${{ matrix.workload }} .json) + echo "driver_name=$DRIVER_NAME" >> $GITHUB_OUTPUT + echo "workload_name=$WORKLOAD_NAME" >> $GITHUB_OUTPUT + echo "result_file=results/github-runner/reference/${DRIVER_NAME}-${WORKLOAD_NAME}.ndjson" >> $GITHUB_OUTPUT + + - name: Run benchmark + run: | + mkdir -p results/github-runner/reference + python -m resp_bench \ + --server localhost:6379 \ + --driver ${{ matrix.driver }} \ + --workload ${{ matrix.workload }} \ + --metrics ${{ steps.names.outputs.result_file }} \ + --commit-id ${{ github.sha }} + + - name: Stop Valkey server + if: always() + run: | + make server-standalone-stop || true + + - name: Upload results + uses: actions/upload-artifact@v4 + with: + name: benchmark-python-${{ steps.names.outputs.driver_name }}-${{ steps.names.outputs.workload_name }} + path: ${{ steps.names.outputs.result_file }} + retention-days: 30 + generate-graphs: - needs: [benchmark-java, benchmark-ruby] + needs: [benchmark-java, benchmark-ruby, benchmark-python] runs-on: ubuntu-latest permissions: contents: write diff --git a/Makefile b/Makefile index db7d5d2..1acd238 100644 --- a/Makefile +++ b/Makefile @@ -31,7 +31,7 @@ WORK_DIR=$(shell pwd)/work server-sentinel-start server-sentinel-stop \ server-start server-stop \ java-build java-test java-run java-clean \ - python-build python-test python-run python-clean \ + python-build python-test python-run python-clean python-info \ ruby-build ruby-test ruby-run ruby-clean ruby-info \ csharp-build csharp-test csharp-run csharp-clean csharp-info \ config-editor-build config-editor-dev @@ -56,10 +56,11 @@ help: @echo " make java-run Run Java benchmark (requires DRIVER and WORKLOAD)" @echo " make java-clean Clean Java build artifacts" @echo "" - @echo "Python Engine (placeholder):" - @echo " make python-build Build Python benchmark engine" + @echo "Python Engine:" + @echo " make python-build Install Python benchmark engine (pip install -e .)" @echo " make python-test Run Python tests" - @echo " make python-run Run Python benchmark" + @echo " make python-run Run Python benchmark (requires DRIVER and WORKLOAD)" + @echo " make python-clean Clean Python build artifacts" @echo "" @echo "Ruby Engine:" @echo " make ruby-build Install Ruby dependencies" @@ -305,24 +306,27 @@ java-info: java-build java -jar $(JAVA_JAR) --info # ============================================================================ -# Python Engine (Placeholder) +# Python Engine # ============================================================================ python-build: - @echo "Python engine not yet implemented" - @echo "Placeholder for: cd python && pip install -e ." + cd python && pip install -e . python-test: - @echo "Python engine not yet implemented" - @echo "Placeholder for: cd python && pytest" + cd python && python -m pytest -python-run: - @echo "Python engine not yet implemented" - @echo "Placeholder for: python -m resp_bench --server $(SERVER) --driver $(DRIVER) --workload $(WORKLOAD)" +python-run: python-build + python -m resp_bench \ + --server $(SERVER) \ + --driver $(DRIVER) \ + --workload $(WORKLOAD) \ + --metrics $(METRICS_OUTPUT) + +python-info: python-build + python -m resp_bench --info python-clean: - @echo "Python engine not yet implemented" - @echo "Placeholder for: cd python && rm -rf __pycache__ *.egg-info dist build" + cd python && rm -rf .venv .pytest_cache __pycache__ dist build *.egg-info src/*.egg-info # ============================================================================ # Ruby Engine diff --git a/configs/drivers/default/redis-py.json b/configs/drivers/default/redis-py.json new file mode 100644 index 0000000..2609cd5 --- /dev/null +++ b/configs/drivers/default/redis-py.json @@ -0,0 +1,8 @@ +{ + "schema_version": "1.0", + "description": "redis-py async client - default configuration", + "driver_id": "redis-py", + "mode": "standalone", + "command_timeout_ms": 5000, + "specific_driver_config": {} +} diff --git a/configs/drivers/default/valkey-glide-python.json b/configs/drivers/default/valkey-glide-python.json new file mode 100644 index 0000000..6495036 --- /dev/null +++ b/configs/drivers/default/valkey-glide-python.json @@ -0,0 +1,8 @@ +{ + "schema_version": "1.0", + "description": "valkey-glide Python async client - default configuration", + "driver_id": "valkey-glide-python", + "mode": "standalone", + "command_timeout_ms": 5000, + "specific_driver_config": {} +} diff --git a/configs/drivers/example-redis-py-standalone.json b/configs/drivers/example-redis-py-standalone.json new file mode 100644 index 0000000..bf759e1 --- /dev/null +++ b/configs/drivers/example-redis-py-standalone.json @@ -0,0 +1,8 @@ +{ + "schema_version": "1.0", + "description": "redis-py async client - standalone mode", + "driver_id": "redis-py", + "mode": "standalone", + "command_timeout_ms": 5000, + "specific_driver_config": {} +} diff --git a/configs/drivers/example-valkey-glide-python-standalone.json b/configs/drivers/example-valkey-glide-python-standalone.json new file mode 100644 index 0000000..4883ae3 --- /dev/null +++ b/configs/drivers/example-valkey-glide-python-standalone.json @@ -0,0 +1,8 @@ +{ + "schema_version": "1.0", + "description": "valkey-glide Python async client - standalone mode", + "driver_id": "valkey-glide-python", + "mode": "standalone", + "command_timeout_ms": 5000, + "specific_driver_config": {} +} diff --git a/configs/drivers/high-throughput/redis-py.json b/configs/drivers/high-throughput/redis-py.json new file mode 100644 index 0000000..979bf81 --- /dev/null +++ b/configs/drivers/high-throughput/redis-py.json @@ -0,0 +1,8 @@ +{ + "schema_version": "1.0", + "description": "redis-py async client - high-throughput configuration", + "driver_id": "redis-py", + "mode": "standalone", + "specific_driver_config": {}, + "command_timeout_ms": 10000 +} diff --git a/configs/drivers/high-throughput/valkey-glide-python.json b/configs/drivers/high-throughput/valkey-glide-python.json new file mode 100644 index 0000000..58aaa9b --- /dev/null +++ b/configs/drivers/high-throughput/valkey-glide-python.json @@ -0,0 +1,8 @@ +{ + "schema_version": "1.0", + "description": "valkey-glide Python async client - high-throughput configuration", + "driver_id": "valkey-glide-python", + "mode": "standalone", + "specific_driver_config": {}, + "command_timeout_ms": 10000 +} diff --git a/docs/ADDING_LANGUAGE.md b/docs/ADDING_LANGUAGE.md index c8adab4..ca59670 100644 --- a/docs/ADDING_LANGUAGE.md +++ b/docs/ADDING_LANGUAGE.md @@ -178,9 +178,10 @@ class MetricsCollector: def record(self, command: str, latency_us: int, success: bool) -> None: if command not in self.command_metrics: - # 1ยตs to 1 hour, 3 significant figures + # 1ยตs to 600s, 3 significant figures (must match the other engines: + # Java/C#/Ruby all use a max of 600_000_000ยตs, not 1 hour) self.command_metrics[command] = CommandMetrics( - histogram=HdrHistogram(1, 3600000000, 3) + histogram=HdrHistogram(1, 600000000, 3) ) metrics = self.command_metrics[command] diff --git a/python/.gitignore b/python/.gitignore new file mode 100644 index 0000000..5fc0ca6 --- /dev/null +++ b/python/.gitignore @@ -0,0 +1,7 @@ +.venv/ +__pycache__/ +*.egg-info/ +*.pyc +dist/ +build/ +.pytest_cache/ diff --git a/python/README.md b/python/README.md index b009bac..ea526c6 100644 --- a/python/README.md +++ b/python/README.md @@ -1,67 +1,79 @@ # resp-bench Python Engine -๐Ÿšง **This engine is planned but not yet implemented.** +Python implementation of the resp-bench benchmark suite, at parity with the +Java (reference), Ruby, and C# engines. -## Overview +## Supported Drivers -Python implementation of the resp-bench benchmark suite. +| Driver | `driver_id` | Package | Notes | +|--------|-------------|---------|-------| +| Valkey GLIDE | `valkey-glide-python` | `valkey-glide` (`import glide`) | Async client | +| redis-py | `redis-py` | `redis` (`redis.asyncio`) | Async client, RESP3, retries disabled | +| Recording | `recording` | โ€” | In-memory; for server-free tests | -## Planned Drivers +Peer drivers are kept deliberately few so each one's configuration can be held +equivalent (same RESP version, same retry policy, same command timeout). A +`valkey-py` driver is a planned follow-up. -| Driver | Package | Status | -|--------|---------|--------| -| redis-py | `redis` | ๐Ÿ“‹ Planned | -| redis-py-async | `redis[hiredis]` | ๐Ÿ“‹ Planned | -| valkey-glide | `valkey-glide` | ๐Ÿ“‹ Planned | +> The GLIDE `driver_id` is `valkey-glide-python` (not the bare `valkey-glide`, +> which is the Java driver) โ€” matching the `valkey-glide-ruby` / +> `valkey-glide-csharp` convention. -## Planned Features +## Execution model -- Full parity with Java engine -- Async/await based execution using `asyncio` -- HdrHistogram for latency collection -- NDJSON metrics output +The engine is asyncio-based. For a phase with `connections = N`, it creates +**N client instances** (one client per connection โ€” the `client == connection` +invariant shared by every engine) and runs **N worker coroutines** concurrently +on a single event loop. Each worker awaits one command at a time, i.e. +`pipeline_depth = 1` โ€” the faithful async analogue of the Java/Ruby +"one in-flight request per connection" model, keeping results comparable across +engines. -## Contributing +This one-client-per-connection mapping is this engine's baseline; it is not a +property of the whole suite (among other engines' drivers, `lettuce` and +`redis-rb` are 1:1, but `jedis`/`redisson` pool, `spring-data-*` share a +template, and `stackexchange-redis` multiplexes). Sharing a single multiplexing client across +workers was proposed and declined upstream +([ikolomi/resp-bench#11](https://github.com/ikolomi/resp-bench/issues/11)) in +favour of keeping this baseline. -We welcome contributions to implement the Python engine! Please see: -- [Architecture Documentation](../docs/ARCHITECTURE.md) -- [Adding a Language Guide](../docs/ADDING_LANGUAGE.md) +### Known limits -## Directory Structure (Planned) +- **`pipeline_depth > 1`** (multiple in-flight requests per connection) is not + implemented; such a phase runs at depth 1 and logs a warning. +- **Single event loop.** Above ~128 connections the event loop, not the driver, + becomes the bottleneck, and loop queuing delay is attributed to the driver in + the reported latency. The engine warns past that threshold. The Java engine hit + the same ceiling with one command-issuing thread and added multiple issuer + threads; this engine has no equivalent yet, so high-connection-count Python + numbers are not directly comparable to other engines. -``` -python/ -โ”œโ”€โ”€ README.md -โ”œโ”€โ”€ pyproject.toml -โ”œโ”€โ”€ requirements.txt -โ””โ”€โ”€ src/ - โ””โ”€โ”€ resp_bench/ - โ”œโ”€โ”€ __init__.py - โ”œโ”€โ”€ __main__.py - โ”œโ”€โ”€ client/ - โ”‚ โ”œโ”€โ”€ __init__.py - โ”‚ โ”œโ”€โ”€ interface.py - โ”‚ โ””โ”€โ”€ impl/ - โ”‚ โ””โ”€โ”€ redis_py.py - โ”œโ”€โ”€ command/ - โ”œโ”€โ”€ config/ - โ”œโ”€โ”€ engine/ - โ””โ”€โ”€ metrics/ -``` - -## Usage (Future) +## Installation ```bash -# Install pip install -e . +# with test tooling: +pip install -e ".[dev]" +``` + +## Usage -# Run benchmark +```bash python -m resp_bench \ --server localhost:6379 \ - --driver ../configs/drivers/example-redis-py-standalone.json \ + --driver ../configs/drivers/default/redis-py.json \ --workload ../configs/workloads/example-workload.json \ --metrics output.ndjson -# Show supported drivers +# Show supported drivers and commands python -m resp_bench --info ``` + +## Testing + +```bash +pytest # unit + recording-driver integration (no server needed) +``` + +See [../docs/ADDING_LANGUAGE.md](../docs/ADDING_LANGUAGE.md) and +[../docs/ARCHITECTURE.md](../docs/ARCHITECTURE.md) for the shared contracts. diff --git a/python/pyproject.toml b/python/pyproject.toml new file mode 100644 index 0000000..29421e8 --- /dev/null +++ b/python/pyproject.toml @@ -0,0 +1,40 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "resp-bench-python" +version = "0.1.0" +description = "Python benchmark engine for resp-bench (async valkey-glide and redis-py)" +readme = "README.md" +requires-python = ">=3.10" +license = { text = "Apache-2.0" } +# Pinned to exact released versions for reproducible benchmark runs, matching the +# Ruby/Java/C# engines. Client-library defaults that affect throughput (RESP +# version, pool size, socket timeouts, retry counts) have changed between minor +# releases, so a range would make results depend on the resolution date. +# hiredis is pinned in deliberately so redis-py always uses its compiled parser +# (GLIDE parses in Rust); the parser actually in use is recorded in the metrics +# metadata. +dependencies = [ + "valkey-glide==2.5.2", + "redis==8.1.0", + "hiredis==3.4.1", + "hdrhistogram==0.10.7", +] + +[project.optional-dependencies] +dev = [ + "pytest>=8.0", + "pytest-asyncio>=0.23", +] + +[project.scripts] +resp-bench = "resp_bench.cli:main" + +[tool.setuptools.packages.find] +where = ["src"] + +[tool.pytest.ini_options] +asyncio_mode = "auto" +testpaths = ["tests"] diff --git a/python/src/resp_bench/__init__.py b/python/src/resp_bench/__init__.py new file mode 100644 index 0000000..db8ca7e --- /dev/null +++ b/python/src/resp_bench/__init__.py @@ -0,0 +1,11 @@ +"""resp-bench Python engine. + +Async implementation of the resp-bench benchmark suite, at parity with the +Java (reference), Ruby, and C# engines. Drives async clients (valkey-glide, +redis-py asyncio) on a single asyncio event loop with one client per +connection. +""" + +from .version import VERSION + +__all__ = ["VERSION"] diff --git a/python/src/resp_bench/__main__.py b/python/src/resp_bench/__main__.py new file mode 100644 index 0000000..be56e85 --- /dev/null +++ b/python/src/resp_bench/__main__.py @@ -0,0 +1,8 @@ +"""Enable ``python -m resp_bench``.""" + +import sys + +from .cli import main + +if __name__ == "__main__": + sys.exit(main()) diff --git a/python/src/resp_bench/cli.py b/python/src/resp_bench/cli.py new file mode 100644 index 0000000..a47e1c3 --- /dev/null +++ b/python/src/resp_bench/cli.py @@ -0,0 +1,126 @@ +"""Command-line interface for the resp-bench Python engine. + +Implements the shared cross-engine CLI contract: ``--server``, ``--driver``, +``--workload``, ``--metrics``, plus ``--info``, ``--commit-id`` (used by CI), +``--version``. (Deliberately no ``--concurrency`` flag: the asyncio engine has +a single execution model.) +""" + +from __future__ import annotations + +import argparse +import asyncio +import logging +import os +import sys +from typing import List, Optional + +from .client.factory import BenchmarkClientFactory +from .command.factory import CommandFactory +from .config.loader import ConfigLoader +from .engine.benchmark import BenchmarkEngine +from .version import VERSION + + +def _build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="resp-bench", + description="resp-bench Python engine", + ) + parser.add_argument("--server", default="localhost:6379", help="Server address HOST:PORT") + parser.add_argument("--driver", help="Driver configuration file (required)") + parser.add_argument("--workload", help="Workload configuration file (required)") + parser.add_argument("--metrics", help="Metrics output file (required)") + parser.add_argument("--commit-id", dest="commit_id", help="Git commit ID for metadata") + parser.add_argument("--info", action="store_true", help="Show supported drivers and commands") + parser.add_argument( + "--version", + action="version", + version=f"resp-bench Python Engine v{VERSION}", + ) + return parser + + +def _parse_server(server: str) -> tuple[str, int]: + host, _, port = server.partition(":") + return host or "localhost", int(port) if port else 6379 + + +def _print_info() -> None: + print(f"resp-bench Python Engine v{VERSION}") + print() + print("Supported Drivers:") + for driver in BenchmarkClientFactory.supported_drivers(): + print(f" - {driver}") + print() + print("Supported Commands:") + for command in CommandFactory.supported_commands(): + print(f" - {command}") + print() + print("Concurrency: asyncio task-per-connection (one client per connection)") + + +def _validate(options: argparse.Namespace) -> None: + missing = [ + flag + for flag, value in ( + ("--driver", options.driver), + ("--workload", options.workload), + ("--metrics", options.metrics), + ) + if not value + ] + if missing: + raise ValueError(f"Missing required options: {', '.join(missing)}") + if not os.path.exists(options.driver): + raise ValueError(f"Driver config not found: {options.driver}") + if not os.path.exists(options.workload): + raise ValueError(f"Workload config not found: {options.workload}") + + +async def _run_benchmark(options: argparse.Namespace) -> bool: + """Run the benchmark; returns True if any phase ended in ERROR.""" + host, port = _parse_server(options.server) + driver_config = ConfigLoader.load_driver_config(options.driver) + workload_config = ConfigLoader.load_workload_config(options.workload) + + engine = BenchmarkEngine( + host=host, + port=port, + driver_config=driver_config, + workload_config=workload_config, + metrics_path=options.metrics, + commit_id=options.commit_id, + ) + await engine.run() + return engine.had_error + + +def main(argv: Optional[List[str]] = None) -> int: + logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") + options = _build_parser().parse_args(argv) + + if options.info: + _print_info() + return 0 + + try: + _validate(options) + had_error = asyncio.run(_run_benchmark(options)) + if had_error: + # Exit non-zero so the matrix runner records the cell as failed + # rather than scoring an unusable run as a good data point. + print("Error: one or more phases ended with status ERROR", file=sys.stderr) + return 1 + return 0 + except Exception as exc: # noqa: BLE001 - top-level CLI error boundary + print(f"Error: {exc}", file=sys.stderr) + if os.environ.get("DEBUG"): + import traceback + + traceback.print_exc() + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/python/src/resp_bench/client/__init__.py b/python/src/resp_bench/client/__init__.py new file mode 100644 index 0000000..3ae7bd7 --- /dev/null +++ b/python/src/resp_bench/client/__init__.py @@ -0,0 +1,7 @@ +"""Async benchmark client interface, driver registry, and implementations.""" + +from .benchmark_client import AsyncBenchmarkClient +from .factory import BenchmarkClientFactory +from .timed_result import TimedResult + +__all__ = ["AsyncBenchmarkClient", "BenchmarkClientFactory", "TimedResult"] diff --git a/python/src/resp_bench/client/benchmark_client.py b/python/src/resp_bench/client/benchmark_client.py new file mode 100644 index 0000000..65b4a86 --- /dev/null +++ b/python/src/resp_bench/client/benchmark_client.py @@ -0,0 +1,80 @@ +"""Abstract async benchmark client. + +Every driver implements this interface. This engine creates **one client +instance per connection** and never shares a client across workers, so a phase's +``connections`` value is also its client count. + +Note this is *this engine's* convention, not a property of the suite: among the +other engines' drivers, ``lettuce`` and ``redis-rb`` map a client to a single +connection, but ``jedis`` uses a pool (``JedisPooled``), ``redisson`` a +connection pool, the ``spring-data-*`` drivers a shared template, and +``stackexchange-redis`` a multiplexer. Sharing one multiplexing client across workers was proposed and +declined upstream (ikolomi/resp-bench#11) in favour of keeping the +one-client-per-connection baseline, which is why this engine does the same. + +Commands are coroutines: a worker awaits one at a time (pipeline depth 1). +""" + +from __future__ import annotations + +import time +from abc import ABC, abstractmethod +from typing import Awaitable, Callable, TypeVar + +from ..config.driver_config import DriverConfig +from .timed_result import TimedResult + +T = TypeVar("T") + + +class AsyncBenchmarkClient(ABC): + @abstractmethod + async def connect(self, host: str, port: int, config: DriverConfig) -> None: + """Establish the connection to the server.""" + + @abstractmethod + async def ping(self) -> TimedResult: + """Execute PING, returning a timed result with value ``PONG``.""" + + @abstractmethod + async def get(self, key: str) -> TimedResult: + """Execute GET, returning a timed result (value or ``None``).""" + + @abstractmethod + async def set(self, key: str, value: bytes) -> TimedResult: + """Execute SET, returning a timed result with value ``OK``.""" + + @abstractmethod + async def close(self) -> None: + """Close the connection.""" + + @abstractmethod + def driver_version(self) -> str: + """Return the underlying driver library version.""" + + def secondary_driver_version(self): # noqa: D401 - optional for composite drivers + """Secondary driver version (composite drivers only).""" + return None + + def driver_details(self) -> dict: + """Environment-dependent settings worth recording in the metrics output. + + Things like the negotiated RESP protocol and the response-parser class + change what is actually being measured, so drivers report them here and + the NDJSON writer records them alongside the driver version. + """ + return {} + + async def _measure(self, operation: Callable[[], Awaitable[T]]) -> TimedResult: + """Await ``operation`` and record its latency in microseconds. + + Latency is captured even on error, matching the other engines. + """ + start = time.perf_counter_ns() + try: + value = await operation() + latency = (time.perf_counter_ns() - start) // 1000 + return TimedResult(value=value, latency_micros=latency) + except Exception as error: # noqa: BLE001 - benchmark records all failures + latency = (time.perf_counter_ns() - start) // 1000 + return TimedResult(value=None, latency_micros=latency, error=error) diff --git a/python/src/resp_bench/client/factory.py b/python/src/resp_bench/client/factory.py new file mode 100644 index 0000000..d048fd7 --- /dev/null +++ b/python/src/resp_bench/client/factory.py @@ -0,0 +1,65 @@ +"""Driver registry: maps ``driver_id`` to a client implementation. + +Implementations are imported lazily so that ``--info`` and unit tests do not +require heavy optional dependencies (valkey-glide, redis) to be installed when +a given driver is not used. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Callable, Dict + +from ..config.driver_config import DriverConfig + +if TYPE_CHECKING: # pragma: no cover + from .benchmark_client import AsyncBenchmarkClient + + +def _make_glide() -> "AsyncBenchmarkClient": + from .impl.glide_client import GlideBenchmarkClient + + return GlideBenchmarkClient() + + +def _make_redis_py() -> "AsyncBenchmarkClient": + from .impl.redis_py_client import RedisPyClient + + return RedisPyClient() + + +def _make_recording() -> "AsyncBenchmarkClient": + from .impl.recording_client import RecordingClient + + return RecordingClient() + + +class BenchmarkClientFactory: + # Ordered so --info lists them predictably. + _FACTORIES: Dict[str, Callable[[], "AsyncBenchmarkClient"]] = { + "valkey-glide-python": _make_glide, + "redis-py": _make_redis_py, + "recording": _make_recording, + } + + @classmethod + def supported_drivers(cls) -> list[str]: + return list(cls._FACTORIES.keys()) + + @classmethod + def create(cls, driver_id: str) -> "AsyncBenchmarkClient": + key = (driver_id or "").lower() + factory = cls._FACTORIES.get(key) + if factory is None: + raise ValueError( + f"Unknown driver: {driver_id}. " + f"Supported: {', '.join(cls._FACTORIES)}" + ) + return factory() + + @classmethod + async def create_and_connect( + cls, host: str, port: int, config: DriverConfig + ) -> "AsyncBenchmarkClient": + client = cls.create(config.driver_id) + await client.connect(host, port, config) + return client diff --git a/python/src/resp_bench/client/impl/__init__.py b/python/src/resp_bench/client/impl/__init__.py new file mode 100644 index 0000000..17dcd10 --- /dev/null +++ b/python/src/resp_bench/client/impl/__init__.py @@ -0,0 +1 @@ +"""Concrete async driver implementations.""" diff --git a/python/src/resp_bench/client/impl/glide_client.py b/python/src/resp_bench/client/impl/glide_client.py new file mode 100644 index 0000000..34ed77f --- /dev/null +++ b/python/src/resp_bench/client/impl/glide_client.py @@ -0,0 +1,82 @@ +"""valkey-glide driver using the async GLIDE Python client (``import glide``). + +The async client is used deliberately (the sync ``glide_sync`` package is not +used). One ``GlideClient`` is created per connection, honoring the +``client == connection`` invariant shared across engines. +""" + +from __future__ import annotations + +from ...config.driver_config import DriverConfig +from ..benchmark_client import AsyncBenchmarkClient +from ..timed_result import TimedResult + + +class GlideBenchmarkClient(AsyncBenchmarkClient): + def __init__(self) -> None: + self._client = None + + async def connect(self, host: str, port: int, config: DriverConfig) -> None: + from glide import ( + GlideClient, + GlideClientConfiguration, + GlideClusterClient, + GlideClusterClientConfiguration, + NodeAddress, + ServerCredentials, + ) + + addresses = [NodeAddress(host, port)] + + credentials = None + if config.auth and (config.auth.get("password") or config.auth.get("username")): + credentials = ServerCredentials( + password=config.auth.get("password", ""), + username=config.auth.get("username"), + ) + + use_tls = config.tls_enabled() + timeout_ms = config.command_timeout_ms + + if config.is_cluster(): + conf = GlideClusterClientConfiguration( + addresses=addresses, + use_tls=use_tls, + credentials=credentials, + request_timeout=timeout_ms, + ) + self._client = await GlideClusterClient.create(conf) + else: + conf = GlideClientConfiguration( + addresses=addresses, + use_tls=use_tls, + credentials=credentials, + request_timeout=timeout_ms, + ) + self._client = await GlideClient.create(conf) + + async def ping(self) -> TimedResult: + return await self._measure(lambda: self._client.ping()) + + async def get(self, key: str) -> TimedResult: + return await self._measure(lambda: self._client.get(key)) + + async def set(self, key: str, value: bytes) -> TimedResult: + return await self._measure(lambda: self._client.set(key, value)) + + async def close(self) -> None: + if self._client is not None: + await self._client.close() + + def driver_version(self) -> str: + try: + from importlib.metadata import version + + return version("valkey-glide") + except Exception: # noqa: BLE001 - version is best-effort metadata + return "unknown" + + def driver_details(self) -> dict: + # GLIDE always negotiates RESP3 and parses in Rust, so neither is + # environment-dependent; recorded for symmetry with the peer drivers. + return {"resp_protocol": 3, "response_parser": "glide-rust", "retries": 0} diff --git a/python/src/resp_bench/client/impl/recording_client.py b/python/src/resp_bench/client/impl/recording_client.py new file mode 100644 index 0000000..618a7c8 --- /dev/null +++ b/python/src/resp_bench/client/impl/recording_client.py @@ -0,0 +1,109 @@ +"""In-memory recording client for server-free testing. + +Records operations and supports simulated latency and error injection via +``specific_driver_config`` (``operation_delay_micros``, +``delay_variation_micros``, ``error_rate``, ``error_message``). This lets the +integration tests exercise the full engine without a live server, mirroring the +Ruby recording driver's role. +""" + +from __future__ import annotations + +import asyncio +import random +import time +from dataclasses import dataclass +from typing import List, Optional + +from ...config.driver_config import DriverConfig +from ..benchmark_client import AsyncBenchmarkClient +from ..timed_result import TimedResult + + +@dataclass +class RecordedOperation: + command: str + key: Optional[str] + value: Optional[bytes] + success: bool + error_message: Optional[str] + + +class RecordingClient(AsyncBenchmarkClient): + def __init__(self) -> None: + self.operations: List[RecordedOperation] = [] + self._stored_data: dict = {} + self._connected = False + self._operation_delay_micros = 0 + self._delay_variation_micros = 0 + self._error_rate = 0.0 + self._error_message = "Simulated error" + self._random = random.Random() + + async def connect(self, host: str, port: int, config: DriverConfig) -> None: + self._connected = True + cfg = config.specific_driver_config or {} + self._operation_delay_micros = int(cfg.get("operation_delay_micros", 0) or 0) + self._delay_variation_micros = int(cfg.get("delay_variation_micros", 0) or 0) + self._error_rate = float(cfg.get("error_rate", 0.0) or 0.0) + self._error_message = str(cfg.get("error_message", "Simulated error")) + self.operations.append(RecordedOperation("CONNECT", None, None, True, None)) + + async def ping(self) -> TimedResult: + latency, success = await self._simulate() + error = None if success else self._error_message + self.operations.append(RecordedOperation("PING", None, None, success, error)) + if success: + return TimedResult(value="PONG", latency_micros=latency) + return TimedResult(value=None, latency_micros=latency, error=RuntimeError(error)) + + async def get(self, key: str) -> TimedResult: + latency, success = await self._simulate() + error = None if success else self._error_message + self.operations.append(RecordedOperation("GET", key, None, success, error)) + if success: + return TimedResult(value=self._stored_data.get(key), latency_micros=latency) + return TimedResult(value=None, latency_micros=latency, error=RuntimeError(error)) + + async def set(self, key: str, value: bytes) -> TimedResult: + latency, success = await self._simulate() + error = None if success else self._error_message + if success: + self._stored_data[key] = value + self.operations.append(RecordedOperation("SET", key, value, success, error)) + if success: + return TimedResult(value="OK", latency_micros=latency) + return TimedResult(value=None, latency_micros=latency, error=RuntimeError(error)) + + async def close(self) -> None: + self._connected = False + self.operations.append(RecordedOperation("CLOSE", None, None, True, None)) + + def driver_version(self) -> str: + return "1.0.0" + + async def _simulate(self) -> tuple[int, bool]: + start = time.perf_counter_ns() + delay_micros = self._calculate_delay_micros() + if delay_micros > 0: + await asyncio.sleep(delay_micros / 1_000_000) + latency = (time.perf_counter_ns() - start) // 1000 + return latency, not self._should_simulate_error() + + def _calculate_delay_micros(self) -> int: + if self._operation_delay_micros <= 0: + return 0 + delay = self._operation_delay_micros + if self._delay_variation_micros > 0: + variation = self._random.randint( + -self._delay_variation_micros, self._delay_variation_micros + ) + delay = max(0, delay + variation) + return delay + + def _should_simulate_error(self) -> bool: + if self._error_rate <= 0.0: + return False + if self._error_rate >= 1.0: + return True + return self._random.random() < self._error_rate diff --git a/python/src/resp_bench/client/impl/redis_py_client.py b/python/src/resp_bench/client/impl/redis_py_client.py new file mode 100644 index 0000000..771c1f4 --- /dev/null +++ b/python/src/resp_bench/client/impl/redis_py_client.py @@ -0,0 +1,109 @@ +"""redis-py driver using the async client (``redis.asyncio``). + +Protocol and retry behaviour are pinned explicitly rather than inherited from the +library defaults, because those defaults changed in redis-py 8.0 (RESP3 by +default, 10 retries with backoff) and differ from other clients'. A benchmark +must measure one round-trip per request, so retries are disabled: a silently +retried failure would otherwise be recorded as a success with an inflated +latency instead of as an error. +""" + +from __future__ import annotations + +from ...config.driver_config import DriverConfig +from ..benchmark_client import AsyncBenchmarkClient +from ..timed_result import TimedResult + + +class RedisPyClient(AsyncBenchmarkClient): + def __init__(self) -> None: + self._client = None + + async def connect(self, host: str, port: int, config: DriverConfig) -> None: + import redis.asyncio as redis_async + + from redis.backoff import NoBackoff + from redis.retry import Retry + + # decode_responses=False keeps values as raw bytes (no decode overhead). + # protocol=3 and retry=0 are set explicitly so the measured behaviour does + # not depend on which redis-py version resolved. + kwargs = { + "host": host, + "port": port, + "decode_responses": False, + "protocol": 3, + "retry": Retry(NoBackoff(), 0), + "retry_on_error": [], + } + + if config.tls_enabled(): + kwargs["ssl"] = True + tls = config.tls or {} + if tls.get("ca_path"): + kwargs["ssl_ca_certs"] = tls["ca_path"] + if tls.get("cert_path"): + kwargs["ssl_certfile"] = tls["cert_path"] + if tls.get("key_path"): + kwargs["ssl_keyfile"] = tls["key_path"] + if tls.get("verify_hostname") is False: + kwargs["ssl_check_hostname"] = False + + if config.auth: + if config.auth.get("username"): + kwargs["username"] = config.auth["username"] + if config.auth.get("password"): + kwargs["password"] = config.auth["password"] + + if config.command_timeout_ms: + kwargs["socket_timeout"] = config.command_timeout_ms / 1000.0 + + if config.is_cluster(): + self._client = redis_async.RedisCluster(**kwargs) + else: + self._client = redis_async.Redis(**kwargs) + + # Establish the connection eagerly so failures surface at connect time. + await self._client.ping() + + async def ping(self) -> TimedResult: + return await self._measure(lambda: self._client.ping()) + + async def get(self, key: str) -> TimedResult: + return await self._measure(lambda: self._client.get(key)) + + async def set(self, key: str, value: bytes) -> TimedResult: + return await self._measure(lambda: self._client.set(key, value)) + + async def close(self) -> None: + if self._client is None: + return + aclose = getattr(self._client, "aclose", None) + if aclose is not None: + await aclose() + else: # pragma: no cover - older redis-py + await self._client.close() + + def driver_version(self) -> str: + import redis + + return getattr(redis, "__version__", "unknown") + + def driver_details(self) -> dict: + """Record the negotiated protocol and the actual parser class. + + Which response parser redis-py picks depends on whether the optional C + extension (hiredis) is importable, so it is recorded rather than assumed. + """ + details = {"resp_protocol": 3, "retries": 0, "response_parser": "unknown"} + try: + # RedisCluster has no connection_pool; it manages pools per node. + pool = getattr(self._client, "connection_pool", None) + if pool is None: + nodes = self._client.nodes_manager.nodes_cache + pool = next(iter(nodes.values())).redis_connection.connection_pool + conn = pool.make_connection() + details["response_parser"] = type(conn._parser).__name__ + except Exception: # noqa: BLE001 - best-effort metadata + pass + return details diff --git a/python/src/resp_bench/client/timed_result.py b/python/src/resp_bench/client/timed_result.py new file mode 100644 index 0000000..ca31634 --- /dev/null +++ b/python/src/resp_bench/client/timed_result.py @@ -0,0 +1,17 @@ +"""Result of a timed client operation.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any, Optional + + +@dataclass +class TimedResult: + value: Any + latency_micros: int + error: Optional[BaseException] = None + + @property + def success(self) -> bool: + return self.error is None diff --git a/python/src/resp_bench/command/__init__.py b/python/src/resp_bench/command/__init__.py new file mode 100644 index 0000000..17c6db3 --- /dev/null +++ b/python/src/resp_bench/command/__init__.py @@ -0,0 +1,6 @@ +"""Benchmark commands and their factory.""" + +from .command import Command, CommandResult +from .factory import CommandFactory + +__all__ = ["Command", "CommandResult", "CommandFactory"] diff --git a/python/src/resp_bench/command/command.py b/python/src/resp_bench/command/command.py new file mode 100644 index 0000000..665ea78 --- /dev/null +++ b/python/src/resp_bench/command/command.py @@ -0,0 +1,28 @@ +"""Command base class and the metrics-facing result struct.""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from dataclasses import dataclass +from typing import Optional + +from ..client.benchmark_client import AsyncBenchmarkClient +from ..config.command_config import CommandConfig + + +@dataclass +class CommandResult: + command_name: str + latency_micros: int + success: bool + + +class Command(ABC): + def __init__(self, config: CommandConfig) -> None: + self.weight = config.weight + self.name = config.command.upper() + self._data_size_bytes = config.data_size_bytes + + @abstractmethod + async def execute(self, client: AsyncBenchmarkClient, key: Optional[str]) -> CommandResult: + """Execute against ``client`` (using ``key`` where applicable).""" diff --git a/python/src/resp_bench/command/factory.py b/python/src/resp_bench/command/factory.py new file mode 100644 index 0000000..80597af --- /dev/null +++ b/python/src/resp_bench/command/factory.py @@ -0,0 +1,38 @@ +"""Factory mapping command names to implementations.""" + +from __future__ import annotations + +from typing import Dict, List, Type + +from ..config.command_config import CommandConfig +from .command import Command +from .impl.get_command import GetCommand +from .impl.ping_command import PingCommand +from .impl.set_command import SetCommand + + +class CommandFactory: + _COMMAND_CLASSES: Dict[str, Type[Command]] = { + "ping": PingCommand, + "get": GetCommand, + "set": SetCommand, + # Future: hget, hset, lpush, lpop, sadd, smembers + } + + @classmethod + def create(cls, config: CommandConfig) -> Command: + command_class = cls._COMMAND_CLASSES.get(config.command) + if command_class is None: + raise ValueError( + f"Unknown command: {config.command}. " + f"Supported: {', '.join(cls._COMMAND_CLASSES)}" + ) + return command_class(config) + + @classmethod + def create_all(cls, configs: List[CommandConfig]) -> List[Command]: + return [cls.create(c) for c in configs] + + @classmethod + def supported_commands(cls) -> List[str]: + return list(cls._COMMAND_CLASSES.keys()) diff --git a/python/src/resp_bench/command/impl/__init__.py b/python/src/resp_bench/command/impl/__init__.py new file mode 100644 index 0000000..79faae7 --- /dev/null +++ b/python/src/resp_bench/command/impl/__init__.py @@ -0,0 +1 @@ +"""Command implementations.""" diff --git a/python/src/resp_bench/command/impl/get_command.py b/python/src/resp_bench/command/impl/get_command.py new file mode 100644 index 0000000..d97d6eb --- /dev/null +++ b/python/src/resp_bench/command/impl/get_command.py @@ -0,0 +1,18 @@ +"""GET command.""" + +from __future__ import annotations + +from typing import Optional + +from ...client.benchmark_client import AsyncBenchmarkClient +from ..command import Command, CommandResult + + +class GetCommand(Command): + async def execute(self, client: AsyncBenchmarkClient, key: Optional[str]) -> CommandResult: + result = await client.get(key) + return CommandResult( + command_name=self.name, + latency_micros=result.latency_micros, + success=result.success, + ) diff --git a/python/src/resp_bench/command/impl/ping_command.py b/python/src/resp_bench/command/impl/ping_command.py new file mode 100644 index 0000000..46f2980 --- /dev/null +++ b/python/src/resp_bench/command/impl/ping_command.py @@ -0,0 +1,18 @@ +"""PING command (ignores the generated key).""" + +from __future__ import annotations + +from typing import Optional + +from ...client.benchmark_client import AsyncBenchmarkClient +from ..command import Command, CommandResult + + +class PingCommand(Command): + async def execute(self, client: AsyncBenchmarkClient, key: Optional[str]) -> CommandResult: + result = await client.ping() + return CommandResult( + command_name=self.name, + latency_micros=result.latency_micros, + success=result.success, + ) diff --git a/python/src/resp_bench/command/impl/set_command.py b/python/src/resp_bench/command/impl/set_command.py new file mode 100644 index 0000000..40875ab --- /dev/null +++ b/python/src/resp_bench/command/impl/set_command.py @@ -0,0 +1,30 @@ +"""SET command with a pre-generated deterministic value.""" + +from __future__ import annotations + +from typing import Optional + +from ...client.benchmark_client import AsyncBenchmarkClient +from ...config.command_config import CommandConfig +from ..command import Command, CommandResult + +_PATTERN = b"0123456789ABCDEF" + + +class SetCommand(Command): + def __init__(self, config: CommandConfig) -> None: + super().__init__(config) + self._value = self._generate_value(self._data_size_bytes) + + async def execute(self, client: AsyncBenchmarkClient, key: Optional[str]) -> CommandResult: + result = await client.set(key, self._value) + return CommandResult( + command_name=self.name, + latency_micros=result.latency_micros, + success=result.success, + ) + + @staticmethod + def _generate_value(size: int) -> bytes: + repeats = (size // len(_PATTERN)) + 1 + return (_PATTERN * repeats)[:size] diff --git a/python/src/resp_bench/config/__init__.py b/python/src/resp_bench/config/__init__.py new file mode 100644 index 0000000..ae24fd4 --- /dev/null +++ b/python/src/resp_bench/config/__init__.py @@ -0,0 +1,19 @@ +"""Configuration models and JSON loader.""" + +from .command_config import CommandConfig +from .completion_config import CompletionConfig +from .driver_config import DriverConfig +from .keyspace_config import KeyspaceConfig +from .loader import ConfigLoader +from .phase_config import PhaseConfig +from .workload_config import WorkloadConfig + +__all__ = [ + "CommandConfig", + "CompletionConfig", + "DriverConfig", + "KeyspaceConfig", + "ConfigLoader", + "PhaseConfig", + "WorkloadConfig", +] diff --git a/python/src/resp_bench/config/command_config.py b/python/src/resp_bench/config/command_config.py new file mode 100644 index 0000000..9402d59 --- /dev/null +++ b/python/src/resp_bench/config/command_config.py @@ -0,0 +1,22 @@ +"""Per-command configuration within a phase.""" + +from __future__ import annotations + +from dataclasses import dataclass + +DEFAULT_DATA_SIZE_BYTES = 256 + + +@dataclass +class CommandConfig: + command: str + weight: float = 1.0 + data_size_bytes: int = DEFAULT_DATA_SIZE_BYTES + + def __post_init__(self) -> None: + self.command = self.command.lower() + # A missing weight defaults to 1.0 (matching the Java reference), rather + # than crashing on float(None). + self.weight = 1.0 if self.weight is None else float(self.weight) + if self.data_size_bytes is None: + self.data_size_bytes = DEFAULT_DATA_SIZE_BYTES diff --git a/python/src/resp_bench/config/completion_config.py b/python/src/resp_bench/config/completion_config.py new file mode 100644 index 0000000..ca69a58 --- /dev/null +++ b/python/src/resp_bench/config/completion_config.py @@ -0,0 +1,58 @@ +"""Phase completion criteria.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Optional + + +DURATION = "duration" +REQUESTS = "requests" + + +@dataclass +class CompletionConfig: + type: str + seconds: Optional[int] = None + requests: Optional[int] = None + + def __post_init__(self) -> None: + # Compared case-insensitively, matching Java (equalsIgnoreCase) and C# + # (OrdinalIgnoreCase) so a config that works on those engines works here. + if isinstance(self.type, str): + self.type = self.type.strip().lower() + + def validate(self) -> None: + """Reject configs that would otherwise run zero requests silently. + + Mirrors the Java reference's CompletionConfig.validate(): the type is + required and must be known, and the relevant bound must be positive. + """ + if not self.type: + raise ValueError("completion.type is required (\"duration\" or \"requests\")") + if self.type == DURATION: + if not self.seconds or self.seconds <= 0: + raise ValueError( + "completion.seconds must be a positive integer for a duration phase" + ) + elif self.type == REQUESTS: + if not self.requests or self.requests <= 0: + raise ValueError( + "completion.requests must be a positive integer for a requests phase" + ) + else: + raise ValueError( + f"Unknown completion type: {self.type} (expected \"duration\" or \"requests\")" + ) + + def is_duration_based(self) -> bool: + return self.type == DURATION + + def is_request_based(self) -> bool: + return self.type == REQUESTS + + def duration_seconds(self) -> int: + return self.seconds if self.seconds is not None else 0 + + def total_requests(self) -> int: + return self.requests if self.requests is not None else 0 diff --git a/python/src/resp_bench/config/driver_config.py b/python/src/resp_bench/config/driver_config.py new file mode 100644 index 0000000..2a13582 --- /dev/null +++ b/python/src/resp_bench/config/driver_config.py @@ -0,0 +1,38 @@ +"""Driver (client library) configuration. + +Maps to configs/schemas/driver-config.schema.json. Field names and defaults +mirror the Ruby/Java engines so the same JSON files work across all engines. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Optional + + +@dataclass +class DriverConfig: + schema_version: str = "1.0" + description: Optional[str] = None + driver_id: Optional[str] = None + mode: str = "standalone" + command_timeout_ms: Optional[int] = None + tls: Optional[dict] = None + auth: Optional[dict] = None + specific_driver_config: dict = field(default_factory=dict) + + def secondary_driver_id(self) -> Optional[Any]: + """Secondary driver id for composite drivers (e.g. spring-data-*).""" + return self.specific_driver_config.get("secondary_driver_id") + + def is_standalone(self) -> bool: + return self.mode == "standalone" + + def is_cluster(self) -> bool: + return self.mode == "cluster" + + def is_sentinel(self) -> bool: + return self.mode == "sentinel" + + def tls_enabled(self) -> bool: + return bool(self.tls and self.tls.get("enabled")) diff --git a/python/src/resp_bench/config/keyspace_config.py b/python/src/resp_bench/config/keyspace_config.py new file mode 100644 index 0000000..eb3fcef --- /dev/null +++ b/python/src/resp_bench/config/keyspace_config.py @@ -0,0 +1,56 @@ +"""Key-generation configuration for a benchmark phase.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Optional + +DEFAULT_KEY_SIZE_BYTES = 16 +DEFAULT_KEY_PREFIX = "bench:" + + +@dataclass +class KeyspaceConfig: + keys_count: int + key_size_bytes: int = DEFAULT_KEY_SIZE_BYTES + key_prefix: str = DEFAULT_KEY_PREFIX + generation_alg: str = "sequential_int" + seed: Optional[int] = None + + def __post_init__(self) -> None: + # Mirror Ruby: nil/None falls back to the default rather than staying None. + if self.key_size_bytes is None: + self.key_size_bytes = DEFAULT_KEY_SIZE_BYTES + if self.key_prefix is None: + self.key_prefix = DEFAULT_KEY_PREFIX + if self.generation_alg is None: + self.generation_alg = "sequential_int" + + def validate(self) -> None: + """Reject a keyspace that would crash the key generator mid-run. + + ``keys_count`` reaches a modulo in KeyGenerator, so 0 or a missing value + would raise ZeroDivisionError/TypeError inside a worker instead of + failing at config load. + """ + if self.keys_count is None or self.keys_count < 1: + raise ValueError("keyspace.keys_count must be a positive integer") + if self.key_size_bytes < 1: + raise ValueError("keyspace.key_size_bytes must be a positive integer") + if self.generation_alg not in ("sequential_int", "uniform_rand"): + raise ValueError( + f"Unknown keyspace.generation_alg: {self.generation_alg} " + '(expected "sequential_int" or "uniform_rand")' + ) + + def is_sequential_int(self) -> bool: + return self.generation_alg == "sequential_int" + + def is_uniform_rand(self) -> bool: + return self.generation_alg == "uniform_rand" + + def effective_key_prefix(self) -> str: + return self.key_prefix if self.key_prefix is not None else DEFAULT_KEY_PREFIX + + def seed_value(self) -> int: + return self.seed if self.seed is not None else 0 diff --git a/python/src/resp_bench/config/loader.py b/python/src/resp_bench/config/loader.py new file mode 100644 index 0000000..20801dc --- /dev/null +++ b/python/src/resp_bench/config/loader.py @@ -0,0 +1,126 @@ +"""Loads driver and workload configuration from JSON files. + +Deserialization mirrors the Ruby ConfigLoader exactly (field names, defaults) +so the shared configs/ JSON files are consumed identically across engines. +""" + +from __future__ import annotations + +import json +from typing import Any, Dict + +from .command_config import CommandConfig +from .completion_config import CompletionConfig +from .driver_config import DriverConfig +from .keyspace_config import KeyspaceConfig +from .phase_config import PhaseConfig +from .workload_config import WorkloadConfig + + +class ConfigLoader: + @staticmethod + def load_driver_config(path: str) -> DriverConfig: + with open(path, encoding="utf-8") as f: + return ConfigLoader.parse_driver_config(json.load(f)) + + @staticmethod + def load_workload_config(path: str) -> WorkloadConfig: + with open(path, encoding="utf-8") as f: + return ConfigLoader.parse_workload_config(json.load(f)) + + @staticmethod + def parse_driver_config(data: Dict[str, Any]) -> DriverConfig: + return DriverConfig( + schema_version=data.get("schema_version", "1.0"), + description=data.get("description"), + driver_id=data.get("driver_id"), + mode=data.get("mode", "standalone"), + command_timeout_ms=data.get("command_timeout_ms"), + tls=data.get("tls"), + auth=data.get("auth"), + specific_driver_config=data.get("specific_driver_config") or {}, + ) + + @staticmethod + def parse_workload_config(data: Dict[str, Any]) -> WorkloadConfig: + phases = [ConfigLoader._parse_phase(p) for p in data.get("phases", [])] + return WorkloadConfig( + schema_version=data.get("schema_version", "1.0"), + benchmark_profile=data.get("benchmark_profile") or {}, + phases=phases, + ) + + @staticmethod + def _parse_phase(data: Dict[str, Any]) -> PhaseConfig: + phase = ConfigLoader._build_phase(data) + # Validate at load time so a bad config fails loudly before any + # connection is opened, rather than running zero requests or crashing a + # worker mid-phase. + phase_id = phase.id or "" + try: + phase.completion.validate() + phase.keyspace.validate() + if phase.connections is None or phase.connections < 1: + raise ValueError("connections must be a positive integer") + if not phase.commands: + raise ValueError("commands must contain at least one entry") + # The shared workload schema permits commands this engine has not + # implemented; without this check they would raise only after every + # connection had been opened. + from ..command.factory import CommandFactory + + supported = CommandFactory.supported_commands() + unsupported = sorted({c.command for c in phase.commands} - set(supported)) + if unsupported: + raise ValueError( + f"unsupported command(s) {', '.join(unsupported)}; " + f"this engine supports: {', '.join(supported)}" + ) + # All-zero weights would make CommandSelector fall through to the last + # command for every pick, silently running a different workload. + if sum(c.weight for c in phase.commands) <= 0: + raise ValueError("command weights must sum to a positive value") + except ValueError as exc: + raise ValueError(f"invalid phase '{phase_id}': {exc}") from exc + return phase + + @staticmethod + def _build_phase(data: Dict[str, Any]) -> PhaseConfig: + return PhaseConfig( + id=data.get("id"), + description=data.get("description"), + connections=data.get("connections"), + cps_limit=data.get("cps_limit", -1), + rps_limit=data.get("rps_limit", -1), + pipeline_depth=data.get("pipeline_depth", 1), + warmup_requests=data.get("warmup_requests", 1), + completion=ConfigLoader._parse_completion(data.get("completion", {})), + keyspace=ConfigLoader._parse_keyspace(data.get("keyspace", {})), + commands=[ConfigLoader._parse_command(c) for c in data.get("commands", [])], + ) + + @staticmethod + def _parse_completion(data: Dict[str, Any]) -> CompletionConfig: + return CompletionConfig( + type=data.get("type"), + seconds=data.get("seconds"), + requests=data.get("requests"), + ) + + @staticmethod + def _parse_keyspace(data: Dict[str, Any]) -> KeyspaceConfig: + return KeyspaceConfig( + keys_count=data.get("keys_count"), + key_size_bytes=data.get("key_size_bytes"), + key_prefix=data.get("key_prefix"), + generation_alg=data.get("generation_alg", "sequential_int"), + seed=data.get("seed"), + ) + + @staticmethod + def _parse_command(data: Dict[str, Any]) -> CommandConfig: + return CommandConfig( + command=data.get("command"), + weight=data.get("weight"), + data_size_bytes=data.get("data_size_bytes"), + ) diff --git a/python/src/resp_bench/config/phase_config.py b/python/src/resp_bench/config/phase_config.py new file mode 100644 index 0000000..cda8189 --- /dev/null +++ b/python/src/resp_bench/config/phase_config.py @@ -0,0 +1,46 @@ +"""Configuration for a single benchmark phase.""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import List, Optional + +from .command_config import CommandConfig +from .completion_config import CompletionConfig +from .keyspace_config import KeyspaceConfig + +DEFAULT_PIPELINE_DEPTH = 1 +DEFAULT_WARMUP_REQUESTS = 1 + + +@dataclass +class PhaseConfig: + id: str + connections: int + completion: CompletionConfig + keyspace: KeyspaceConfig + commands: List[CommandConfig] + description: Optional[str] = None + cps_limit: int = -1 + rps_limit: int = -1 + pipeline_depth: int = DEFAULT_PIPELINE_DEPTH + warmup_requests: int = DEFAULT_WARMUP_REQUESTS + + def __post_init__(self) -> None: + if self.cps_limit is None: + self.cps_limit = -1 + if self.rps_limit is None: + self.rps_limit = -1 + if self.pipeline_depth is None: + self.pipeline_depth = DEFAULT_PIPELINE_DEPTH + if self.warmup_requests is None: + self.warmup_requests = DEFAULT_WARMUP_REQUESTS + + def has_cps_limit(self) -> bool: + return self.cps_limit > 0 + + def has_rps_limit(self) -> bool: + return self.rps_limit > 0 + + def effective_pipeline_depth(self) -> int: + return self.pipeline_depth if self.pipeline_depth > 0 else DEFAULT_PIPELINE_DEPTH diff --git a/python/src/resp_bench/config/workload_config.py b/python/src/resp_bench/config/workload_config.py new file mode 100644 index 0000000..3c6de62 --- /dev/null +++ b/python/src/resp_bench/config/workload_config.py @@ -0,0 +1,21 @@ +"""Configuration for a benchmark workload (a sequence of phases).""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import List, Optional + +from .phase_config import PhaseConfig + + +@dataclass +class WorkloadConfig: + schema_version: str + benchmark_profile: dict = field(default_factory=dict) + phases: List[PhaseConfig] = field(default_factory=list) + + def name(self) -> Optional[str]: + return self.benchmark_profile.get("name") + + def description(self) -> Optional[str]: + return self.benchmark_profile.get("description") diff --git a/python/src/resp_bench/engine/__init__.py b/python/src/resp_bench/engine/__init__.py new file mode 100644 index 0000000..4586524 --- /dev/null +++ b/python/src/resp_bench/engine/__init__.py @@ -0,0 +1 @@ +"""Benchmark execution engine and its deterministic building blocks.""" diff --git a/python/src/resp_bench/engine/benchmark.py b/python/src/resp_bench/engine/benchmark.py new file mode 100644 index 0000000..f25eac8 --- /dev/null +++ b/python/src/resp_bench/engine/benchmark.py @@ -0,0 +1,410 @@ +"""Async benchmark engine. + +Concurrency model: a single asyncio event loop with **one client per +connection** and **one worker coroutine per connection**, run concurrently. +Each worker awaits one command at a time -- i.e. ``pipeline_depth`` is +effectively 1, the same "one in-flight request per connection" shape the Java +engine's virtual-thread workers produce. + +A request-based phase target is a single budget shared across all workers, which +they claim from one request at a time -- matching the Java reference's shared +``AtomicLong`` rather than pre-splitting the target per worker. + +Known limits of this model, both deliberate: + +* **Single event loop.** Above roughly ``HIGH_CONNECTION_WARN_THRESHOLD`` + connections the loop itself, not the driver, becomes the bottleneck, and the + loop's queuing delay is attributed to the driver in the reported latency. The + Java engine faced the same ceiling with a single command-issuing thread (see + ``docs/ARCHITECTURE.md``) and solved it with multiple issuer threads; this + engine has no equivalent yet, so it warns above the threshold. Do not compare + high-connection-count Python results against other engines without accounting + for this. +* **``pipeline_depth > 1``** (multiple in-flight requests per connection) is not + implemented; a phase requesting it runs at depth 1 with a warning. +""" + +from __future__ import annotations + +import asyncio +import logging +import time +from typing import List, Optional + +from ..client.benchmark_client import AsyncBenchmarkClient +from ..client.factory import BenchmarkClientFactory +from ..command.command import Command, CommandResult +from ..command.factory import CommandFactory +from ..config.driver_config import DriverConfig +from ..config.phase_config import PhaseConfig +from ..config.workload_config import WorkloadConfig +from ..metrics.collector import MetricsCollector +from ..metrics.ndjson_writer import NdjsonWriter +from .command_selector import CommandSelector +from .key_generator import Counter, KeyGenerator +from .rate_limiter import RateLimiter + +logger = logging.getLogger("resp_bench") + +# Above this many connections the single event loop, not the driver, tends to set +# the throughput ceiling and its queuing delay shows up as driver latency. +HIGH_CONNECTION_WARN_THRESHOLD = 128 + +# A worker yields at least this often even when every request completes without +# suspending (a cache hit, or an in-memory driver). Without it one connection can +# monopolise a whole duration-based phase while its peers record nothing. +YIELD_EVERY_N_REQUESTS = 64 + +# Backoff applied after repeated consecutive failures on one connection, so a +# permanently-failing connection cannot spin at full CPU inflating the error +# count. Capped so a transient blip costs almost nothing. +CONSECUTIVE_FAILURE_BACKOFF_AFTER = 8 +MAX_FAILURE_BACKOFF_SECONDS = 0.05 + + +async def _gather_all_or_cancel(coros) -> None: + """Await all coroutines; on the first failure cancel and drain the rest. + + ``asyncio.gather`` deliberately does NOT cancel siblings when one task + raises, which would leave workers running against clients the caller has + already closed. This wrapper cancels them and waits for them to finish + unwinding before re-raising, so no work escapes the phase that started it. + """ + tasks = [asyncio.ensure_future(c) for c in coros] + try: + await asyncio.gather(*tasks) + except BaseException: + for task in tasks: + if not task.done(): + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + raise + + +class BenchmarkEngine: + def __init__( + self, + *, + host: str, + port: int, + driver_config: DriverConfig, + workload_config: WorkloadConfig, + metrics_path: str, + commit_id: Optional[str] = None, + ) -> None: + self._host = host + self._port = port + self._driver_config = driver_config + self._workload_config = workload_config + self._writer = NdjsonWriter(metrics_path) + self._commit_id = commit_id + # True if any phase ended in ERROR. The CLI turns this into a non-zero + # exit code so the matrix runner does not score the cell as a good run. + self._had_error = False + + @property + def had_error(self) -> bool: + return self._had_error + + async def run(self) -> None: + logger.info("Starting benchmark: %s", self._workload_config.name()) + logger.info( + "Driver: %s, Server mode: %s", + self._driver_config.driver_id, + self._driver_config.mode, + ) + logger.info("Concurrency: asyncio task-per-connection (one client per connection)") + logger.info("Server: %s:%s", self._host, self._port) + + await self._setup_metadata() + + for phase in self._workload_config.phases: + await self._execute_phase(phase) + + logger.info("Benchmark completed") + + async def _setup_metadata(self) -> None: + try: + sample = await BenchmarkClientFactory.create_and_connect( + self._host, self._port, self._driver_config + ) + self._writer.set_metadata( + commit_id=self._commit_id, + driver_id=self._driver_config.driver_id, + primary_driver_version=sample.driver_version(), + secondary_driver_id=self._driver_config.secondary_driver_id(), + secondary_driver_version=sample.secondary_driver_version(), + driver_details=sample.driver_details(), + ) + logger.info( + "Metadata: commit=%s, driver=%s, version=%s, details=%s", + self._commit_id or "N/A", + self._driver_config.driver_id, + sample.driver_version(), + sample.driver_details() or "{}", + ) + await sample.close() + except Exception as exc: # noqa: BLE001 - metadata is best-effort + logger.warning("Failed to get driver version for metadata: %s", exc) + self._writer.set_metadata( + commit_id=self._commit_id, + driver_id=self._driver_config.driver_id, + primary_driver_version="unknown", + secondary_driver_id=self._driver_config.secondary_driver_id(), + secondary_driver_version=None, + ) + + async def _execute_phase(self, phase: PhaseConfig) -> None: + logger.info("=== Starting phase: %s (%s) ===", phase.id, phase.description) + + if phase.effective_pipeline_depth() > 1: + logger.warning( + "pipeline_depth=%d requested for phase '%s', but the Python engine " + "does not yet implement pipelining; running at depth 1. Results are " + "not comparable to pipelined runs of other engines.", + phase.pipeline_depth, + phase.id, + ) + + if phase.connections > HIGH_CONNECTION_WARN_THRESHOLD: + logger.warning( + "connections=%d exceeds %d: the single event loop is likely the " + "bottleneck rather than the driver, and loop queuing delay is " + "reported as driver latency. Treat these results with care and do " + "not compare them directly against other engines.", + phase.connections, + HIGH_CONNECTION_WARN_THRESHOLD, + ) + + collector = MetricsCollector() + status = "ERROR" + failure: Optional[BaseException] = None + clients: List[AsyncBenchmarkClient] = [] + try: + # Inside the guarded region: a failure part-way through opening + # connections (server down, maxclients reached) or an unsupported + # command name must still close what was opened and emit a row. + clients = await self._create_clients(phase) + commands = CommandFactory.create_all(phase.commands) + + if phase.warmup_requests > 0: + await self._warmup(clients, phase.warmup_requests) + + # Started here so the measured window covers the workload only, not + # connection setup or warmup. + collector.start() + status = await self._run_workload(phase, clients, commands, collector) + except Exception as exc: # noqa: BLE001 - recorded below, then re-raised + # Failures used to escape before anything was written, so the phase + # produced no row at all. Always emit a row so it stays visible. + failure = exc + logger.error("Phase %s failed: %s", phase.id, exc) + finally: + # A phase that failed before the workload started still needs real + # timestamps: nulls would violate the documented schema and the graph + # scripts would aggregate the row as a 0-RPS data point. + if collector.start_time is None: + collector.start() + collector.stop() + await self._close_clients(clients) + + if status == "ERROR": + self._had_error = True + + self._writer.write_phase_results( + phase_id=phase.id, + status=status, + connections=phase.connections, + collector=collector, + ) + self._log_phase_summary(phase, collector, status) + + if failure is not None: + raise failure + + async def _create_clients(self, phase: PhaseConfig) -> List[AsyncBenchmarkClient]: + logger.info("Creating %d connections...", phase.connections) + cps_limiter = RateLimiter.create(phase.cps_limit) if phase.has_cps_limit() else None + + clients: List[AsyncBenchmarkClient] = [] + for _ in range(phase.connections): + if cps_limiter is not None: + await cps_limiter.acquire() + client = await BenchmarkClientFactory.create_and_connect( + self._host, self._port, self._driver_config + ) + clients.append(client) + logger.info("All %d connections established", len(clients)) + return clients + + async def _warmup(self, clients: List[AsyncBenchmarkClient], warmup_requests: int) -> None: + logger.info("Warmup: %d PINGs per client...", warmup_requests) + + async def warm(client: AsyncBenchmarkClient) -> None: + for _ in range(warmup_requests): + result = await client.ping() + # Fail fast on an unreachable/misconfigured server rather than + # running a whole phase that records only errors. + if not result.success: + raise RuntimeError(f"Warmup PING failed: {result.error}") + + # Cancel-and-drain on first failure so no warmup task keeps pinging a + # client that _execute_phase has already closed. + await _gather_all_or_cancel(warm(c) for c in clients) + logger.info("Warmup completed") + + async def _run_workload( + self, + phase: PhaseConfig, + clients: List[AsyncBenchmarkClient], + commands: List[Command], + collector: MetricsCollector, + ) -> str: + completion = phase.completion + num_workers = len(clients) + seed_base = phase.keyspace.seed_value() + shared_counter = Counter() # shared across workers for sequential_int + + # Constructed here, at the start of the measured window, rather than + # before warmup: the limiter's schedule starts from its construction + # time, so building it earlier would let the phase open with a burst of + # accumulated slots proportional to the warmup duration. + rate_limiter = RateLimiter.create(phase.rps_limit) if phase.has_rps_limit() else None + + # A request-based target is a single budget SHARED across workers, which + # each worker claims from one request at a time (matching the Java + # reference's shared AtomicLong). A slow connection therefore cannot cap + # the phase -- faster workers absorb the slack and the phase ends when + # the total budget is exhausted. Duration-based runs use a wall-clock + # deadline instead. + target_requests = None if completion.is_duration_based() else completion.total_requests() + end_time = ( + time.monotonic() + completion.duration_seconds() + if completion.is_duration_based() + else None + ) + request_budget = Counter() + + async def worker(idx: int, client: AsyncBenchmarkClient) -> None: + key_gen = KeyGenerator.create_with_seed( + phase.keyspace, seed_base + idx, sequential_counter=shared_counter + ) + selector = CommandSelector(commands) + since_yield = 0 + consecutive_failures = 0 + while True: + if target_requests is not None: + # Claim a slot; claims are atomic on the single event loop + # (no await between read and increment), so unlike Java no + # decrement-on-overshoot is needed. + if request_budget.next_value() >= target_requests: + break + elif time.monotonic() >= end_time: + break + + if rate_limiter is not None: + await rate_limiter.acquire() + + command = None + succeeded = False + try: + command = selector.select() + key = key_gen.next_key() + result = await command.execute(client, key) + collector.record(result) + succeeded = result.success + except Exception: # noqa: BLE001 - record failures, keep going + collector.record( + CommandResult( + command_name=command.name if command else "UNKNOWN", + latency_micros=0, + success=False, + ) + ) + + # A request can complete without ever suspending -- a driver that + # raises before its first await (caught by _measure), or any + # in-memory/cache-hit response. Such a worker would monopolise the + # event loop, so guarantee a suspension point. + since_yield += 1 + if succeeded: + consecutive_failures = 0 + if since_yield >= YIELD_EVERY_N_REQUESTS: + since_yield = 0 + await asyncio.sleep(0) + else: + consecutive_failures += 1 + since_yield = 0 + if consecutive_failures >= CONSECUTIVE_FAILURE_BACKOFF_AFTER: + # Back off so a permanently-failing connection cannot spin + # at full CPU inflating the error count. + await asyncio.sleep( + min( + 0.001 * (consecutive_failures - CONSECUTIVE_FAILURE_BACKOFF_AFTER + 1), + MAX_FAILURE_BACKOFF_SECONDS, + ) + ) + else: + await asyncio.sleep(0) + + logger.info("Starting %d worker coroutines...", num_workers) + try: + # Cancel-and-drain on first failure: a worker that dies must not + # leave its peers running against clients we are about to close. + await _gather_all_or_cancel(worker(i, c) for i, c in enumerate(clients)) + except KeyboardInterrupt: # pragma: no cover + logger.warning("Workload interrupted") + return "INTERRUPTED" + except Exception as exc: # noqa: BLE001 + logger.error("Error during workload execution: %s", exc) + return "ERROR" + + # A phase in which nothing succeeded produced no usable latency data; + # reporting COMPLETED would let the orchestrator record it as a good run. + successes = collector.total_requests - collector.total_errors + if collector.total_requests > 0 and successes == 0: + logger.error( + "All %d requests failed; reporting phase as ERROR", + collector.total_requests, + ) + return "ERROR" + + logger.info("All operations completed (%d total requests)", collector.total_requests) + return "COMPLETED" + + async def _close_clients(self, clients: List[AsyncBenchmarkClient]) -> None: + logger.info("Closing %d connections...", len(clients)) + for client in clients: + try: + await client.close() + except Exception as exc: # noqa: BLE001 + logger.warning("Error closing client: %s", exc) + + def _log_phase_summary(self, phase, collector, status) -> None: + duration_s = collector.duration_millis() / 1000.0 + total = collector.total_requests + errors = collector.total_errors + rps = round(total / duration_s) if duration_s > 0 else 0 + logger.info("=== Phase %s completed: %s ===", phase.id, status) + logger.info( + " Duration: %.1fs | Requests: %d | Errors: %d | RPS: %d", + duration_s, + total, + errors, + rps, + ) + for cmd_name, cmd_metrics in collector.command_metrics.items(): + if cmd_metrics.count() == 0: + continue + logger.info( + " %s: %d req (%d err) | p50=%dus p95=%dus p99=%dus p99.9=%dus | min=%dus max=%dus", + cmd_name, + cmd_metrics.requests, + cmd_metrics.errors, + cmd_metrics.percentile(50), + cmd_metrics.percentile(95), + cmd_metrics.percentile(99), + cmd_metrics.percentile(99.9), + cmd_metrics.min(), + cmd_metrics.max(), + ) diff --git a/python/src/resp_bench/engine/command_selector.py b/python/src/resp_bench/engine/command_selector.py new file mode 100644 index 0000000..9bbc31d --- /dev/null +++ b/python/src/resp_bench/engine/command_selector.py @@ -0,0 +1,40 @@ +"""Weighted command selection. + +Uses normalized cumulative weights. Command selection intentionally uses +Python's built-in RNG (not the Java LCG): only key generation must be +cross-engine deterministic, command selection need not be. +""" + +from __future__ import annotations + +import random +from typing import List + +from ..command.command import Command + + +class CommandSelector: + def __init__(self, commands: List[Command]) -> None: + self._commands = commands + self._cumulative_weights = self._build_cumulative_weights(commands) + self._random = random.Random() + + def select(self) -> Command: + r = self._random.random() + for index, threshold in enumerate(self._cumulative_weights): + if r <= threshold: + return self._commands[index] + return self._commands[-1] + + @staticmethod + def _build_cumulative_weights(commands: List[Command]) -> List[float]: + total_weight = sum(c.weight for c in commands) + if total_weight == 0: + total_weight = 1.0 + + cumulative: List[float] = [] + running = 0.0 + for command in commands: + running += command.weight / total_weight + cumulative.append(running) + return cumulative diff --git a/python/src/resp_bench/engine/java_random.py b/python/src/resp_bench/engine/java_random.py new file mode 100644 index 0000000..1c61f1e --- /dev/null +++ b/python/src/resp_bench/engine/java_random.py @@ -0,0 +1,57 @@ +"""Faithful port of ``java.util.Random`` (48-bit LCG). + +This guarantees identical ``uniform_rand`` key sequences across the Java +(reference), Ruby, C#, and Python engines. Java is the canonical source, so +this port reproduces ``nextInt(bound)`` exactly -- including the 32-bit +signed-overflow rejection in the general case, which is what avoids modulo +bias. (Note: the Ruby port omits that rejection because Ruby integers are +arbitrary-precision; this port emulates the int32 wraparound so it matches the +Java reference rather than the Ruby approximation.) + +@see https://docs.oracle.com/javase/8/docs/api/java/util/Random.html +""" + +from __future__ import annotations + +MULTIPLIER = 0x5DEECE66D +ADDEND = 0xB +MASK = (1 << 48) - 1 + + +def _to_int32(value: int) -> int: + """Interpret the low 32 bits of ``value`` as a signed 32-bit integer.""" + value &= 0xFFFFFFFF + return value - 0x100000000 if value >= 0x80000000 else value + + +class JavaRandom: + def __init__(self, seed: int) -> None: + self._seed = self._initial_scramble(seed) + + def set_seed(self, seed: int) -> None: + self._seed = self._initial_scramble(seed) + + def next_int(self, bound: int) -> int: + """Return a random int in ``[0, bound)`` matching Java's nextInt(int).""" + if bound <= 0: + raise ValueError("bound must be positive") + + # Power-of-two fast path (matches Java exactly). + if (bound & -bound) == bound: + return (bound * self._next_bits(31)) >> 31 + + # General case: rejection sampling to avoid modulo bias. The rejection + # condition relies on 32-bit signed overflow, which we emulate. + while True: + bits = self._next_bits(31) + val = bits % bound + if _to_int32(bits - val + (bound - 1)) >= 0: + return val + + @staticmethod + def _initial_scramble(seed: int) -> int: + return (seed ^ MULTIPLIER) & MASK + + def _next_bits(self, bits: int) -> int: + self._seed = ((self._seed * MULTIPLIER) + ADDEND) & MASK + return self._seed >> (48 - bits) diff --git a/python/src/resp_bench/engine/key_generator.py b/python/src/resp_bench/engine/key_generator.py new file mode 100644 index 0000000..f60f3b1 --- /dev/null +++ b/python/src/resp_bench/engine/key_generator.py @@ -0,0 +1,88 @@ +"""Key generator producing sequences identical to the other engines. + +- ``sequential_int``: keys 0, 1, 2, ... N-1, wrapping around. +- ``uniform_rand``: Java-LCG random keys (see :mod:`.java_random`). + +Key formatting matches Java's ``String.format("%0Nd", index)``: the numeric +part is zero-padded to ``max(key_size_bytes - len(prefix), 1)`` digits. + +Cross-worker semantics follow the Java reference: +- ``sequential_int`` uses a counter SHARED across all workers in a phase, so + the workers collectively emit 0, 1, 2, ... (this is what populates the whole + keyspace during a warmup/populate phase). Pass a shared :class:`Counter`. +- ``uniform_rand`` uses a per-worker RNG seeded ``base_seed + worker_index``. +""" + +from __future__ import annotations + +from typing import Optional + +from ..config.keyspace_config import KeyspaceConfig +from .java_random import JavaRandom + + +class Counter: + """A monotonic 0-based counter. Shared across a phase's workers. + + Safe to share across asyncio tasks: ``next_value`` performs its + read-increment with no ``await`` in between, so it is atomic on the single + event-loop thread. + """ + + def __init__(self, start: int = 0) -> None: + self._value = start + + def next_value(self) -> int: + current = self._value + self._value += 1 + return current + + def reset(self) -> None: + self._value = 0 + + +class KeyGenerator: + def __init__( + self, + config: KeyspaceConfig, + seed_override: Optional[int] = None, + sequential_counter: Optional[Counter] = None, + ) -> None: + self._config = config + self._key_prefix = config.effective_key_prefix() + self._key_size_bytes = config.key_size_bytes + self._keys_count = config.keys_count + self._seed = seed_override if seed_override is not None else config.seed_value() + self._sequential_counter = sequential_counter or Counter() + self._random = JavaRandom(self._seed) + + @classmethod + def create(cls, config: KeyspaceConfig) -> "KeyGenerator": + return cls(config) + + @classmethod + def create_with_seed( + cls, + config: KeyspaceConfig, + seed: int, + sequential_counter: Optional[Counter] = None, + ) -> "KeyGenerator": + """Per-worker generator with a unique seed and an optional shared counter.""" + return cls(config, seed_override=seed, sequential_counter=sequential_counter) + + def next_key(self) -> str: + if self._config.is_sequential_int(): + key_index = self._sequential_counter.next_value() + else: + key_index = self._random.next_int(self._keys_count) + + key_index %= self._keys_count + return self._format_key(key_index) + + def reset(self) -> None: + self._sequential_counter.reset() + self._random.set_seed(self._seed) + + def _format_key(self, key_index: int) -> str: + padding_width = max(self._key_size_bytes - len(self._key_prefix), 1) + return f"{self._key_prefix}{key_index:0{padding_width}d}" diff --git a/python/src/resp_bench/engine/rate_limiter.py b/python/src/resp_bench/engine/rate_limiter.py new file mode 100644 index 0000000..4283a74 --- /dev/null +++ b/python/src/resp_bench/engine/rate_limiter.py @@ -0,0 +1,41 @@ +"""Async leaky-bucket rate limiter. + +Enforces a constant rate with no burst (evenly-spaced operations), matching the +Java reference's interval math: ``interval_ns = 1_000_000_000 // rate``. Unlike +the blocking engines, ``acquire`` yields the event loop via ``asyncio.sleep`` +so other connections' coroutines make progress while this one waits. + +A single limiter is shared across all of a phase's worker coroutines. Because +the event loop is single-threaded, the check-and-advance of ``next_allowed`` +has no ``await`` between read and write, so it is atomic (no CAS needed). +""" + +from __future__ import annotations + +import asyncio +import time +from typing import Optional + + +class RateLimiter: + def __init__(self, rate_per_second: int) -> None: + self.rate_per_second = rate_per_second + self._interval_nanos = 1_000_000_000 // rate_per_second + self._next_allowed_nanos = time.monotonic_ns() + + @staticmethod + def create(rate_per_second: int) -> Optional["RateLimiter"]: + """Return a limiter, or ``None`` for unlimited (rate <= 0).""" + if rate_per_second <= 0: + return None + return RateLimiter(rate_per_second) + + async def acquire(self) -> None: + while True: + now = time.monotonic_ns() + if now >= self._next_allowed_nanos: + self._next_allowed_nanos += self._interval_nanos + return + wait_seconds = (self._next_allowed_nanos - now) / 1_000_000_000 + if wait_seconds > 0: + await asyncio.sleep(wait_seconds) diff --git a/python/src/resp_bench/metrics/__init__.py b/python/src/resp_bench/metrics/__init__.py new file mode 100644 index 0000000..1f0636d --- /dev/null +++ b/python/src/resp_bench/metrics/__init__.py @@ -0,0 +1,6 @@ +"""Latency collection and NDJSON output.""" + +from .collector import CommandMetrics, MetricsCollector +from .ndjson_writer import NdjsonWriter + +__all__ = ["CommandMetrics", "MetricsCollector", "NdjsonWriter"] diff --git a/python/src/resp_bench/metrics/collector.py b/python/src/resp_bench/metrics/collector.py new file mode 100644 index 0000000..a0e1980 --- /dev/null +++ b/python/src/resp_bench/metrics/collector.py @@ -0,0 +1,81 @@ +"""Latency metrics collection. + +Single-event-loop design: no locks are needed because ``record`` runs to +completion without awaiting, so concurrent worker coroutines never interleave +inside it. Latencies are clamped to 600s before recording and errors are +counted but not recorded into the histogram -- matching the other engines. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Dict, Optional + +from ..command.command import CommandResult +from .hdr_encoder import HIGHEST_TRACKABLE_VALUE, new_histogram + + +class CommandMetrics: + def __init__(self, command_name: str) -> None: + self.command_name = command_name + self.requests = 0 + self.errors = 0 + # Created eagerly (like the Java reference) so the NDJSON hdr block and + # summary are always present, even for a command that only ever errors + # (an empty histogram reports count 0 and zero percentiles). + self._histogram = new_histogram() + + def record(self, result: CommandResult) -> None: + self.requests += 1 + if result.success: + latency = min(result.latency_micros, HIGHEST_TRACKABLE_VALUE) + self._histogram.record_value(latency) + else: + self.errors += 1 + + @property + def histogram(self): + return self._histogram + + def count(self) -> int: + return self._histogram.get_total_count() + + def min(self) -> int: + return self._histogram.get_min_value() + + def max(self) -> int: + return self._histogram.get_max_value() + + def percentile(self, pct: float) -> int: + return self._histogram.get_value_at_percentile(pct) + + +class MetricsCollector: + def __init__(self) -> None: + self.command_metrics: Dict[str, CommandMetrics] = {} + self.total_requests = 0 + self.total_errors = 0 + self.start_time: Optional[datetime] = None + self.end_time: Optional[datetime] = None + + def start(self) -> None: + self.start_time = datetime.now(timezone.utc) + + def stop(self) -> None: + self.end_time = datetime.now(timezone.utc) + + def record(self, result: CommandResult) -> None: + self.total_requests += 1 + if not result.success: + self.total_errors += 1 + + metrics = self.command_metrics.get(result.command_name) + if metrics is None: + metrics = CommandMetrics(result.command_name) + self.command_metrics[result.command_name] = metrics + metrics.record(result) + + def duration_millis(self) -> int: + if not self.start_time or not self.end_time: + return 0 + return int((self.end_time - self.start_time).total_seconds() * 1000) diff --git a/python/src/resp_bench/metrics/hdr_encoder.py b/python/src/resp_bench/metrics/hdr_encoder.py new file mode 100644 index 0000000..4277322 --- /dev/null +++ b/python/src/resp_bench/metrics/hdr_encoder.py @@ -0,0 +1,37 @@ +"""HdrHistogram helpers. + +Uses the ``hdrhistogram`` PyPI package (import ``hdrh``), the official Python +port of HdrHistogram. Its ``encode()`` emits the base64-encoded V2 *compressed* +payload -- the same format Java's ``encodeIntoCompressedByteBuffer`` produces +and the Ruby engine emits -- so payloads are mutually decodable across engines +for cross-language analysis. (Byte-identity is not guaranteed because zlib +compression levels may differ, but decodability -- what merge/analysis needs -- +is.) + +Histograms use range ``(1, 600_000_000, 3)``: 1 microsecond to 600 seconds at 3 +significant figures, matching every other engine (Java +``SynchronizedHistogram(600_000_000, 3)``, C# ``LongConcurrentHistogram(1, +600_000_000, 3)``, Ruby ``HDRHistogram.new(1, 600_000_000, 3)``). +""" + +from __future__ import annotations + +from hdrh.histogram import HdrHistogram + +LOWEST_TRACKABLE_VALUE = 1 +HIGHEST_TRACKABLE_VALUE = 600_000_000 # 600 seconds in microseconds +SIGNIFICANT_FIGURES = 3 + + +def new_histogram() -> HdrHistogram: + return HdrHistogram( + LOWEST_TRACKABLE_VALUE, HIGHEST_TRACKABLE_VALUE, SIGNIFICANT_FIGURES + ) + + +def encode_base64(histogram: HdrHistogram) -> str: + """Return the base64 V2-compressed encoding as an ASCII string.""" + encoded = histogram.encode() + if isinstance(encoded, bytes): + return encoded.decode("ascii") + return encoded diff --git a/python/src/resp_bench/metrics/ndjson_writer.py b/python/src/resp_bench/metrics/ndjson_writer.py new file mode 100644 index 0000000..ee7d3ce --- /dev/null +++ b/python/src/resp_bench/metrics/ndjson_writer.py @@ -0,0 +1,135 @@ +"""Writes benchmark metrics as NDJSON (one JSON object per phase). + +The output schema matches every other engine exactly (see +docs/CONFIG_SPECIFICATION.md): ``metadata`` / ``phase`` / ``totals`` / +``metrics`` blocks, latency ``unit: "us"``, integer ``summary`` percentiles, +uppercased command keys, and an HDR block with the base64 compressed payload. +""" + +from __future__ import annotations + +import json +import os +from datetime import datetime, timezone +from typing import Optional + +from .collector import MetricsCollector +from .hdr_encoder import encode_base64 + + +def _iso8601_utc(dt: Optional[datetime]) -> Optional[str]: + if dt is None: + return None + return dt.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") + + +class NdjsonWriter: + def __init__(self, path: str) -> None: + self._output_path = path + self._commit_id = None + self._driver_id = None + self._primary_driver_version = None + self._secondary_driver_id = None + self._secondary_driver_version = None + self._driver_details = {} + + def set_metadata( + self, + *, + commit_id: Optional[str], + driver_id: Optional[str], + primary_driver_version: Optional[str], + secondary_driver_id: Optional[str] = None, + secondary_driver_version: Optional[str] = None, + driver_details: Optional[dict] = None, + ) -> None: + self._commit_id = commit_id + self._driver_id = driver_id + self._primary_driver_version = primary_driver_version + self._secondary_driver_id = secondary_driver_id + self._secondary_driver_version = secondary_driver_version + self._driver_details = driver_details or {} + + def write_phase_results( + self, + *, + phase_id: str, + status: str, + connections: int, + collector: MetricsCollector, + ) -> None: + parent = os.path.dirname(self._output_path) + if parent: + os.makedirs(parent, exist_ok=True) + + payload = self._build_phase_json(phase_id, status, connections, collector) + with open(self._output_path, "a", encoding="utf-8") as f: + f.write(json.dumps(payload) + "\n") + + def _build_phase_json(self, phase_id, status, connections, collector) -> dict: + result: dict = {} + + if self._commit_id or self._driver_id: + metadata: dict = {} + if self._commit_id: + metadata["commit_id"] = self._commit_id + metadata["timestamp"] = _iso8601_utc(datetime.now(timezone.utc)) + if self._driver_id: + metadata["driver_id"] = self._driver_id + if self._primary_driver_version: + metadata["primary_driver_version"] = self._primary_driver_version + if self._secondary_driver_id: + metadata["secondary_driver_id"] = self._secondary_driver_id + if self._secondary_driver_version: + metadata["secondary_driver_version"] = self._secondary_driver_version + # Additive, optional fields: settings that change what is actually + # being measured (negotiated protocol, response parser, retry count). + # Other engines omit them, and downstream tooling reads metadata by + # key, so extra keys are ignored there. + for key, value in self._driver_details.items(): + metadata.setdefault(key, value) + result["metadata"] = metadata + + result["phase"] = { + "id": phase_id, + "status": status, + "start_timestamp": _iso8601_utc(collector.start_time), + "finish_timestamp": _iso8601_utc(collector.end_time), + "duration_ms": collector.duration_millis(), + "connections": connections, + } + + result["totals"] = { + "requests": collector.total_requests, + "errors": collector.total_errors, + } + + result["metrics"] = self._build_command_metrics(collector) + return result + + def _build_command_metrics(self, collector: MetricsCollector) -> dict: + metrics: dict = {} + for cmd_name, cmd_metrics in collector.command_metrics.items(): + latency: dict = { + "unit": "us", + "count": cmd_metrics.count(), + "summary": { + "min": int(cmd_metrics.min()), + "p50": int(cmd_metrics.percentile(50)), + "p95": int(cmd_metrics.percentile(95)), + "p99": int(cmd_metrics.percentile(99)), + "p999": int(cmd_metrics.percentile(99.9)), + "max": int(cmd_metrics.max()), + }, + } + latency["hdr"] = { + "format": "hdr", + "sigfig": 3, + "payload_b64": encode_base64(cmd_metrics.histogram), + } + metrics[cmd_name] = { + "requests": cmd_metrics.requests, + "errors": cmd_metrics.errors, + "latency": latency, + } + return metrics diff --git a/python/src/resp_bench/version.py b/python/src/resp_bench/version.py new file mode 100644 index 0000000..9429261 --- /dev/null +++ b/python/src/resp_bench/version.py @@ -0,0 +1,3 @@ +"""Engine version.""" + +VERSION = "0.1.0" diff --git a/python/tests/integration/test_recording_workload.py b/python/tests/integration/test_recording_workload.py new file mode 100644 index 0000000..7cc9985 --- /dev/null +++ b/python/tests/integration/test_recording_workload.py @@ -0,0 +1,131 @@ +"""End-to-end engine tests using the recording driver (no server needed).""" + +import json + +import pytest + +from resp_bench.config.command_config import CommandConfig +from resp_bench.config.completion_config import CompletionConfig +from resp_bench.config.driver_config import DriverConfig +from resp_bench.config.keyspace_config import KeyspaceConfig +from resp_bench.config.phase_config import PhaseConfig +from resp_bench.config.workload_config import WorkloadConfig +from resp_bench.engine.benchmark import BenchmarkEngine + + +def _driver(**specific): + return DriverConfig(driver_id="recording", specific_driver_config=specific) + + +def _workload(phases): + return WorkloadConfig(schema_version="1.0", benchmark_profile={"name": "t"}, phases=phases) + + +def _phase(**kw): + defaults = dict( + id="P", + connections=4, + completion=CompletionConfig(type="requests", requests=400), + keyspace=KeyspaceConfig(keys_count=100, key_prefix="k:", generation_alg="sequential_int"), + commands=[CommandConfig(command="set", weight=1.0, data_size_bytes=64)], + warmup_requests=0, + ) + defaults.update(kw) + return PhaseConfig(**defaults) + + +async def _run(tmp_path, driver, workload): + out = tmp_path / "metrics.ndjson" + engine = BenchmarkEngine( + host="localhost", + port=6379, + driver_config=driver, + workload_config=workload, + metrics_path=str(out), + ) + await engine.run() + return [json.loads(line) for line in out.read_text().splitlines()] + + +async def test_request_based_completion_totals(tmp_path): + rows = await _run(tmp_path, _driver(), _workload([_phase()])) + assert len(rows) == 1 + row = rows[0] + assert row["phase"]["status"] == "COMPLETED" + assert row["phase"]["connections"] == 4 + assert row["totals"]["requests"] == 400 + assert row["totals"]["errors"] == 0 + assert row["metrics"]["SET"]["requests"] == 400 + assert row["metrics"]["SET"]["errors"] == 0 + + +async def test_shared_budget_hits_exact_total_when_uneven(tmp_path): + # The request target is a single shared budget (matching Java), not split + # per worker, so a target that does not divide evenly across connections + # must still produce exactly that many requests. + rows = await _run( + tmp_path, + _driver(), + _workload( + [_phase(connections=4, completion=CompletionConfig(type="requests", requests=401))] + ), + ) + assert rows[0]["totals"]["requests"] == 401 + assert rows[0]["metrics"]["SET"]["requests"] == 401 + + +async def test_error_injection_counts_errors(tmp_path): + rows = await _run( + tmp_path, + _driver(error_rate=1.0), + _workload([_phase(completion=CompletionConfig(type="requests", requests=100))]), + ) + row = rows[0] + assert row["totals"]["requests"] == 100 + assert row["totals"]["errors"] == 100 + # All failed -> empty (but present) histogram: count 0, zero summary, and + # the hdr block is still emitted (matching the Java reference schema). + set_metrics = row["metrics"]["SET"] + assert set_metrics["errors"] == 100 + assert set_metrics["latency"]["count"] == 0 + assert set_metrics["latency"]["summary"]["max"] == 0 + assert set_metrics["latency"]["hdr"]["format"] == "hdr" + + +async def test_warmup_fails_fast_on_all_errors(tmp_path): + # A server that fails every request should abort at warmup rather than + # running a whole phase of error metrics. + out = tmp_path / "metrics.ndjson" + engine = BenchmarkEngine( + host="localhost", + port=6379, + driver_config=_driver(error_rate=1.0), + workload_config=_workload([_phase(warmup_requests=2)]), + metrics_path=str(out), + ) + with pytest.raises(RuntimeError, match="Warmup"): + await engine.run() + # The failure still produces a row, marked ERROR, so the phase is visible in + # the metrics output instead of silently vanishing. + row = json.loads(out.read_text().splitlines()[0]) + assert row["phase"]["status"] == "ERROR" + assert row["totals"]["requests"] == 0 + + +async def test_two_phases_written_as_two_lines(tmp_path): + phases = [ + _phase(id="WARMUP", completion=CompletionConfig(type="requests", requests=100)), + _phase( + id="STEADY", + completion=CompletionConfig(type="requests", requests=200), + commands=[ + CommandConfig(command="get", weight=0.8), + CommandConfig(command="set", weight=0.2, data_size_bytes=64), + ], + warmup_requests=1, + ), + ] + rows = await _run(tmp_path, _driver(), _workload(phases)) + assert [r["phase"]["id"] for r in rows] == ["WARMUP", "STEADY"] + assert rows[0]["totals"]["requests"] == 100 + assert rows[1]["totals"]["requests"] == 200 diff --git a/python/tests/integration/test_worker_isolation.py b/python/tests/integration/test_worker_isolation.py new file mode 100644 index 0000000..8e34abd --- /dev/null +++ b/python/tests/integration/test_worker_isolation.py @@ -0,0 +1,249 @@ +"""Worker lifecycle: no orphans, no loop starvation, honest status. + +Covers the three failure modes a reviewer reproduced on the original engine: +a dying worker leaving its peers running past client close, a synchronously +failing driver monopolising the event loop, and a warmup failure producing no +metrics row at all. +""" + +import asyncio +import json + +import pytest + +from resp_bench.client import factory +from resp_bench.client.impl.recording_client import RecordingClient +from resp_bench.config.command_config import CommandConfig +from resp_bench.config.completion_config import CompletionConfig +from resp_bench.config.driver_config import DriverConfig +from resp_bench.config.keyspace_config import KeyspaceConfig +from resp_bench.config.phase_config import PhaseConfig +from resp_bench.config.workload_config import WorkloadConfig +from resp_bench.engine.benchmark import BenchmarkEngine + + +@pytest.fixture(autouse=True) +def _restore_factory(): + original = factory.BenchmarkClientFactory._FACTORIES["recording"] + yield + factory.BenchmarkClientFactory._FACTORIES["recording"] = original + + +def _engine(tmp_path, phase, name="metrics.ndjson"): + out = tmp_path / name + return out, BenchmarkEngine( + host="localhost", + port=6379, + driver_config=DriverConfig(driver_id="recording", specific_driver_config={}), + workload_config=WorkloadConfig( + schema_version="1.0", benchmark_profile={"name": "t"}, phases=[phase] + ), + metrics_path=str(out), + ) + + +def _phase(**kw): + defaults = dict( + id="P", + connections=4, + completion=CompletionConfig(type="requests", requests=2000), + keyspace=KeyspaceConfig(keys_count=50, key_prefix="k:", generation_alg="sequential_int"), + commands=[CommandConfig(command="set", weight=1.0, data_size_bytes=32)], + warmup_requests=0, + ) + defaults.update(kw) + return PhaseConfig(**defaults) + + +async def test_gather_all_or_cancel_cancels_siblings(): + # asyncio.gather alone leaves siblings running when one task raises, which is + # what let workers keep issuing against closed clients. + from resp_bench.engine.benchmark import _gather_all_or_cancel + + completed = [] + + async def slow(tag): + await asyncio.sleep(0.5) + completed.append(tag) + + async def boom(): + await asyncio.sleep(0) + raise RuntimeError("injected") + + with pytest.raises(RuntimeError, match="injected"): + await _gather_all_or_cancel([slow("a"), boom(), slow("b")]) + + # Siblings were cancelled and drained before the exception surfaced. + assert completed == [] + pending = [t for t in asyncio.all_tasks() if t is not asyncio.current_task() and not t.done()] + assert pending == [] + + +async def test_driver_failure_is_recorded_not_fatal_and_leaves_no_orphans(tmp_path): + made = [] + + class Dying(RecordingClient): + async def connect(self, host, port, config): + await super().connect(host, port, config) + made.append(self) + self.calls = 0 + self.post_close = 0 + self._operation_delay_micros = 200 + + async def set(self, key, value): + self.calls += 1 + if not self._connected: + self.post_close += 1 + # made[0] is the metadata probe; made[1] is the first worker. + if len(made) > 1 and self is made[1] and self.calls % 5 == 0: + raise RuntimeError("injected driver failure") + return await super().set(key, value) + + factory.BenchmarkClientFactory._FACTORIES["recording"] = lambda: Dying() + out, engine = _engine(tmp_path, _phase(completion=CompletionConfig(type="requests", requests=400))) + await engine.run() + + # No worker task outlived run(), and nothing issued after close. + pending = [t for t in asyncio.all_tasks() if t is not asyncio.current_task() and not t.done()] + assert pending == [] + await asyncio.sleep(0.05) + assert sum(c.post_close for c in made) == 0, "a worker kept issuing after close" + + row = json.loads(out.read_text().splitlines()[0]) + # The failure is accounted as an error rather than killing the phase, and the + # full budget is still honoured. + assert row["totals"]["requests"] == 400 + assert row["totals"]["errors"] > 0 + assert row["phase"]["status"] == "COMPLETED" + + +async def test_synchronously_failing_driver_does_not_starve_the_loop(tmp_path): + made = [] + + class SyncFail(RecordingClient): + async def connect(self, host, port, config): + await super().connect(host, port, config) + made.append(self) + self.ops = 0 + # Healthy peers take real (awaited) time; the failing one returns + # instantly without ever suspending. + self._operation_delay_micros = 200 + + async def set(self, key, value): + self.ops += 1 + if len(made) > 1 and self is made[1]: + raise RuntimeError("fails before awaiting anything") + return await super().set(key, value) + + factory.BenchmarkClientFactory._FACTORIES["recording"] = lambda: SyncFail() + out, engine = _engine( + tmp_path, _phase(connections=4, completion=CompletionConfig(type="requests", requests=400)) + ) + await engine.run() + + healthy = [c.ops for c in made[2:]] + # Before the fix the synchronously-failing worker never suspended and the + # healthy connections completed zero requests. + assert all(n > 0 for n in healthy), f"healthy connections were starved: {healthy}" + row = json.loads(out.read_text().splitlines()[0]) + assert row["totals"]["requests"] == 400 + assert row["metrics"]["SET"]["latency"]["count"] > 0, "no successful latencies recorded" + + +async def test_all_requests_failing_reports_error_not_completed(tmp_path): + out, engine = _engine( + tmp_path, + _phase(completion=CompletionConfig(type="requests", requests=200)), + ) + engine._driver_config.specific_driver_config = {"error_rate": 1.0} + await engine.run() + + row = json.loads(out.read_text().splitlines()[0]) + assert row["totals"]["requests"] == 200 + assert row["totals"]["errors"] == 200 + # A phase with no successful request produced no usable latency data. + assert row["phase"]["status"] == "ERROR" + + +async def test_zero_latency_driver_still_shares_work_across_connections(tmp_path): + # A driver that completes without suspending (in-memory, or a cache hit) must + # not let one connection monopolise a duration-based phase. + made = [] + + class Counting(RecordingClient): + async def connect(self, host, port, config): + await super().connect(host, port, config) + made.append(self) + self.ops = 0 + + async def set(self, key, value): + self.ops += 1 + return await super().set(key, value) + + factory.BenchmarkClientFactory._FACTORIES["recording"] = lambda: Counting() + # operation_delay_micros defaults to 0 -> the driver never suspends. + out, engine = _engine( + tmp_path, + _phase(connections=4, completion=CompletionConfig(type="duration", seconds=1)), + ) + await engine.run() + + per_conn = [c.ops for c in made[1:]] + assert all(n > 0 for n in per_conn), f"one connection monopolised the phase: {per_conn}" + row = json.loads(out.read_text().splitlines()[0]) + assert row["phase"]["status"] == "COMPLETED" + + +async def test_failure_before_workload_still_writes_row_with_timestamps(tmp_path): + # Connection setup failing must close what was opened and still emit a row -- + # with real timestamps, since nulls violate the schema and the graph scripts + # would aggregate the row as a 0-RPS point. + class NoConnect(RecordingClient): + async def connect(self, host, port, config): + raise ConnectionError("server unreachable") + + factory.BenchmarkClientFactory._FACTORIES["recording"] = lambda: NoConnect() + out, engine = _engine(tmp_path, _phase()) + + with pytest.raises(ConnectionError): + await engine.run() + + row = json.loads(out.read_text().splitlines()[0]) + assert row["phase"]["status"] == "ERROR" + assert row["phase"]["start_timestamp"] is not None + assert row["phase"]["finish_timestamp"] is not None + assert engine.had_error is True + + +async def test_warmup_failure_still_writes_a_row_and_cancels_peers(tmp_path): + made = [] + + class WarmFail(RecordingClient): + async def connect(self, host, port, config): + await super().connect(host, port, config) + made.append(self) + self.pings = 0 + + async def ping(self): + self.pings += 1 + if len(made) > 1 and self is made[1]: + from resp_bench.client.timed_result import TimedResult + + return TimedResult(value=None, latency_micros=1, error=RuntimeError("boom")) + return await super().ping() + + factory.BenchmarkClientFactory._FACTORIES["recording"] = lambda: WarmFail() + out, engine = _engine(tmp_path, _phase(warmup_requests=200)) + # Give the healthy peers a real suspension point so cancellation is observable + # (a zero-latency in-memory client would run all 200 pings in one slice). + engine._driver_config.specific_driver_config = {"operation_delay_micros": 200} + + with pytest.raises(RuntimeError, match="Warmup"): + await engine.run() + + # The phase must still be represented in the output rather than vanishing. + row = json.loads(out.read_text().splitlines()[0]) + assert row["phase"]["status"] == "ERROR" + # Peers were cancelled rather than left pinging closed clients. + healthy_pings = [c.pings for c in made[2:]] + assert all(p < 200 for p in healthy_pings), f"peers ran to completion: {healthy_pings}" diff --git a/python/tests/unit/test_command_selector.py b/python/tests/unit/test_command_selector.py new file mode 100644 index 0000000..e66fcc7 --- /dev/null +++ b/python/tests/unit/test_command_selector.py @@ -0,0 +1,27 @@ +from resp_bench.command.factory import CommandFactory +from resp_bench.config.command_config import CommandConfig +from resp_bench.engine.command_selector import CommandSelector + + +def test_respects_weights_approximately(): + commands = CommandFactory.create_all( + [ + CommandConfig(command="get", weight=0.8), + CommandConfig(command="set", weight=0.2, data_size_bytes=64), + ] + ) + selector = CommandSelector(commands) + + counts = {"GET": 0, "SET": 0} + n = 20000 + for _ in range(n): + counts[selector.select().name] += 1 + + get_fraction = counts["GET"] / n + assert 0.75 <= get_fraction <= 0.85 + + +def test_single_command_always_selected(): + commands = CommandFactory.create_all([CommandConfig(command="ping", weight=1.0)]) + selector = CommandSelector(commands) + assert all(selector.select().name == "PING" for _ in range(100)) diff --git a/python/tests/unit/test_config_loader.py b/python/tests/unit/test_config_loader.py new file mode 100644 index 0000000..704c8dc --- /dev/null +++ b/python/tests/unit/test_config_loader.py @@ -0,0 +1,90 @@ +from pathlib import Path + +from resp_bench.config.loader import ConfigLoader + +# repo root: tests/unit/test_x.py -> parents[3] +REPO_ROOT = Path(__file__).resolve().parents[3] +CONFIGS = REPO_ROOT / "configs" + + +def test_parse_driver_config_defaults(): + cfg = ConfigLoader.parse_driver_config({"driver_id": "redis-py"}) + assert cfg.schema_version == "1.0" + assert cfg.mode == "standalone" + assert cfg.specific_driver_config == {} + assert cfg.command_timeout_ms is None + assert cfg.is_standalone() + + +def test_parse_driver_config_top_level_command_timeout(): + # command_timeout_ms is a top-level key (as in the shared config files), + # not nested under specific_driver_config. + cfg = ConfigLoader.parse_driver_config( + {"driver_id": "redis-py", "command_timeout_ms": 10000} + ) + assert cfg.command_timeout_ms == 10000 + + +def test_command_without_weight_defaults_to_one(): + workload = ConfigLoader.parse_workload_config( + { + "schema_version": "1.0", + "benchmark_profile": {"name": "t"}, + "phases": [ + { + "id": "P", + "connections": 1, + "completion": {"type": "requests", "requests": 1}, + "keyspace": {"keys_count": 1, "key_prefix": "k:", "generation_alg": "sequential_int"}, + "commands": [{"command": "ping"}], + } + ], + } + ) + assert workload.phases[0].commands[0].weight == 1.0 + + +def test_parse_phase_defaults(): + workload = ConfigLoader.parse_workload_config( + { + "schema_version": "1.0", + "benchmark_profile": {"name": "t"}, + "phases": [ + { + "id": "P", + "connections": 2, + "completion": {"type": "requests", "requests": 10}, + "keyspace": {"keys_count": 5, "key_prefix": "k:", "generation_alg": "sequential_int"}, + "commands": [{"command": "SET", "weight": 1.0}], + } + ], + } + ) + phase = workload.phases[0] + assert phase.cps_limit == -1 + assert phase.rps_limit == -1 + assert phase.pipeline_depth == 1 + assert phase.warmup_requests == 1 + # command name is lowercased at config level; keyspace defaults applied. + assert phase.commands[0].command == "set" + assert phase.commands[0].data_size_bytes == 256 + assert phase.keyspace.key_size_bytes == 16 + + +def test_loads_shared_example_workload(): + workload = ConfigLoader.load_workload_config( + str(CONFIGS / "workloads" / "example-workload.json") + ) + assert len(workload.phases) == 2 + warmup, steady = workload.phases + assert warmup.id == "WARMUP" + assert warmup.completion.is_request_based() + assert steady.keyspace.is_uniform_rand() + assert steady.keyspace.seed_value() == 12345 + + +def test_loads_shared_driver_configs(): + for name in ("redis-rb.json", "valkey-glide.json"): + cfg = ConfigLoader.load_driver_config(str(CONFIGS / "drivers" / "default" / name)) + assert cfg.schema_version == "1.0" + assert cfg.driver_id diff --git a/python/tests/unit/test_config_validation.py b/python/tests/unit/test_config_validation.py new file mode 100644 index 0000000..f3e09bd --- /dev/null +++ b/python/tests/unit/test_config_validation.py @@ -0,0 +1,111 @@ +"""Config validation: reject configs that would silently run zero work. + +Ported from the Java reference's CompletionConfig.validate(), plus keyspace +validation so an unusable keyspace fails at load instead of crashing a worker. +""" + +import pytest + +from resp_bench.config.loader import ConfigLoader + + +def _workload(completion, keyspace=None, connections=1, commands=None): + return { + "schema_version": "1.0", + "benchmark_profile": {"name": "t"}, + "phases": [ + { + "id": "P", + "connections": connections, + "completion": completion, + "keyspace": keyspace + or {"keys_count": 10, "key_prefix": "k:", "generation_alg": "sequential_int"}, + "commands": commands if commands is not None else [{"command": "get", "weight": 1.0}], + } + ], + } + + +@pytest.mark.parametrize( + "completion, expected", + [ + ({"type": "seconds", "seconds": 2}, "Unknown completion type"), + ({"type": "requests"}, "completion.requests must be a positive integer"), + ({"type": "duration"}, "completion.seconds must be a positive integer"), + ({"type": "requests", "requests": 0}, "completion.requests must be a positive integer"), + ({"type": "duration", "seconds": 0}, "completion.seconds must be a positive integer"), + ({}, "completion.type is required"), + ], +) +def test_invalid_completion_rejected(completion, expected): + with pytest.raises(ValueError, match=expected): + ConfigLoader.parse_workload_config(_workload(completion)) + + +@pytest.mark.parametrize("type_str", ["duration", "Duration", "DURATION", " duration "]) +def test_completion_type_is_case_insensitive(type_str): + # Java compares with equalsIgnoreCase and C# with OrdinalIgnoreCase, so a + # config that works there must work here rather than silently running 0 requests. + wl = ConfigLoader.parse_workload_config(_workload({"type": type_str, "seconds": 2})) + phase = wl.phases[0] + assert phase.completion.is_duration_based() + assert phase.completion.duration_seconds() == 2 + + +@pytest.mark.parametrize( + "keyspace, expected", + [ + ({"keys_count": 0, "key_prefix": "k:", "generation_alg": "sequential_int"}, + "keys_count must be a positive integer"), + ({"key_prefix": "k:", "generation_alg": "sequential_int"}, + "keys_count must be a positive integer"), + ({"keys_count": 10, "key_prefix": "k:", "generation_alg": "shuffle"}, + "Unknown keyspace.generation_alg"), + ], +) +def test_invalid_keyspace_rejected(keyspace, expected): + with pytest.raises(ValueError, match=expected): + ConfigLoader.parse_workload_config( + _workload({"type": "requests", "requests": 10}, keyspace=keyspace) + ) + + +def test_invalid_connections_and_commands_rejected(): + with pytest.raises(ValueError, match="connections must be a positive integer"): + ConfigLoader.parse_workload_config( + _workload({"type": "requests", "requests": 10}, connections=0) + ) + with pytest.raises(ValueError, match="commands must contain at least one entry"): + ConfigLoader.parse_workload_config( + _workload({"type": "requests", "requests": 10}, commands=[]) + ) + + +def test_unsupported_command_rejected_at_load(): + # The shared workload schema allows these; this engine has not implemented + # them, and without validation they would raise only after every connection + # had been opened. + with pytest.raises(ValueError, match="unsupported command"): + ConfigLoader.parse_workload_config( + _workload( + {"type": "requests", "requests": 10}, + commands=[{"command": "hset", "weight": 1.0}], + ) + ) + + +def test_zero_total_weight_rejected(): + # All-zero weights make CommandSelector fall through to the last command for + # every pick, silently running a different workload than configured. + with pytest.raises(ValueError, match="weights must sum to a positive value"): + ConfigLoader.parse_workload_config( + _workload( + {"type": "requests", "requests": 10}, + commands=[{"command": "get", "weight": 0}, {"command": "set", "weight": 0}], + ) + ) + + +def test_error_names_the_phase(): + with pytest.raises(ValueError, match="invalid phase 'P'"): + ConfigLoader.parse_workload_config(_workload({"type": "nope"})) diff --git a/python/tests/unit/test_factory.py b/python/tests/unit/test_factory.py new file mode 100644 index 0000000..cec109b --- /dev/null +++ b/python/tests/unit/test_factory.py @@ -0,0 +1,24 @@ +import pytest + +from resp_bench.client.factory import BenchmarkClientFactory +from resp_bench.command.factory import CommandFactory + + +def test_supported_drivers(): + drivers = BenchmarkClientFactory.supported_drivers() + assert drivers == ["valkey-glide-python", "redis-py", "recording"] + + +def test_create_recording_driver(): + # The recording driver needs no optional deps and must instantiate. + client = BenchmarkClientFactory.create("recording") + assert client.driver_version() == "1.0.0" + + +def test_create_unknown_driver_raises(): + with pytest.raises(ValueError, match="Unknown driver"): + BenchmarkClientFactory.create("nope") + + +def test_supported_commands(): + assert CommandFactory.supported_commands() == ["ping", "get", "set"] diff --git a/python/tests/unit/test_hdr_encoder.py b/python/tests/unit/test_hdr_encoder.py new file mode 100644 index 0000000..89e88b7 --- /dev/null +++ b/python/tests/unit/test_hdr_encoder.py @@ -0,0 +1,34 @@ +from hdrh.histogram import HdrHistogram + +from resp_bench.metrics.hdr_encoder import ( + HIGHEST_TRACKABLE_VALUE, + LOWEST_TRACKABLE_VALUE, + SIGNIFICANT_FIGURES, + encode_base64, + new_histogram, +) + + +def test_histogram_range_matches_other_engines(): + assert (LOWEST_TRACKABLE_VALUE, HIGHEST_TRACKABLE_VALUE, SIGNIFICANT_FIGURES) == ( + 1, + 600_000_000, + 3, + ) + + +def test_encode_is_base64_v2_compressed_and_decodes(): + h = new_histogram() + for v in (5, 50, 500, 5000, 50000): + h.record_value(v) + + b64 = encode_base64(h) + assert isinstance(b64, str) + # HdrHistogram base64 of the compressed V2 payload begins with "HIST" + # (the compressed cookie 0x1c849314) -- same family as the Ruby/Java output. + assert b64.startswith("HIST") + + decoded = HdrHistogram.decode(b64.encode("ascii")) + assert decoded.get_total_count() == 5 + assert decoded.get_value_at_percentile(50) == h.get_value_at_percentile(50) + assert decoded.get_max_value() == h.get_max_value() diff --git a/python/tests/unit/test_java_random.py b/python/tests/unit/test_java_random.py new file mode 100644 index 0000000..4806e0d --- /dev/null +++ b/python/tests/unit/test_java_random.py @@ -0,0 +1,59 @@ +"""JavaRandom parity tests. + +The core LCG is anchored to a well-known java.util.Random value, which proves +byte-for-byte compatibility with the Java reference without needing a JVM. +""" + +import pytest + +from resp_bench.engine.java_random import JavaRandom, _to_int32 + + +def test_seed_zero_matches_known_java_value(): + # java.util.Random(0).nextInt() (i.e. next(32) as a signed int) is the + # well-documented value -1155484576. This anchors the LCG to real Java. + rng = JavaRandom(0) + assert _to_int32(rng._next_bits(32)) == -1155484576 + + +def test_deterministic_sequence(): + a = [JavaRandom(12345).next_int(1000) for _ in range(10)] + b = [JavaRandom(12345).next_int(1000) for _ in range(10)] + assert a == b + + +def test_different_seeds_differ(): + a = [JavaRandom(12345).next_int(1000) for _ in range(10)] + b = [JavaRandom(54321).next_int(1000) for _ in range(10)] + assert a != b + + +def test_set_seed_resets(): + rng = JavaRandom(12345) + first = [rng.next_int(1000) for _ in range(5)] + rng.set_seed(12345) + second = [rng.next_int(1000) for _ in range(5)] + assert first == second + + +def test_bound_must_be_positive(): + rng = JavaRandom(12345) + with pytest.raises(ValueError): + rng.next_int(0) + with pytest.raises(ValueError): + rng.next_int(-1) + + +def test_values_within_bound(): + rng = JavaRandom(12345) + for _ in range(1000): + v = rng.next_int(100) + assert 0 <= v < 100 + + +def test_power_of_two_bounds(): + rng = JavaRandom(12345) + for bound in (2, 4, 8, 16, 256, 1024): + for _ in range(200): + v = rng.next_int(bound) + assert 0 <= v < bound diff --git a/python/tests/unit/test_key_generator.py b/python/tests/unit/test_key_generator.py new file mode 100644 index 0000000..56aa6ea --- /dev/null +++ b/python/tests/unit/test_key_generator.py @@ -0,0 +1,61 @@ +from resp_bench.config.keyspace_config import KeyspaceConfig +from resp_bench.engine.key_generator import Counter, KeyGenerator + + +def _keyspace(**kw): + defaults = dict(keys_count=100, key_prefix="test:", generation_alg="sequential_int") + defaults.update(kw) + return KeyspaceConfig(**defaults) + + +def test_sequential_wraps_around(): + gen = KeyGenerator.create(_keyspace(keys_count=3)) + keys = [gen.next_key() for _ in range(6)] + assert keys[0] == keys[3] + assert keys[1] == keys[4] + assert keys[2] == keys[5] + assert keys[0] != keys[1] + + +def test_sequential_shared_counter_across_workers(): + # Java shares the sequential counter across a phase's workers so they + # collectively emit 0, 1, 2, ... (populating the whole keyspace). + counter = Counter() + g0 = KeyGenerator.create_with_seed(_keyspace(keys_count=1000), 0, sequential_counter=counter) + g1 = KeyGenerator.create_with_seed(_keyspace(keys_count=1000), 1, sequential_counter=counter) + keys = [] + for _ in range(5): + keys.append(g0.next_key()) + keys.append(g1.next_key()) + # 10 draws from a shared counter -> 10 distinct keys (indices 0..9). + assert len(set(keys)) == 10 + + +def test_uniform_rand_reproducible(): + ks = _keyspace(keys_count=1000, key_prefix="rand:", generation_alg="uniform_rand", seed=12345) + a = [KeyGenerator.create(ks).next_key() for _ in range(20)] + ks2 = _keyspace(keys_count=1000, key_prefix="rand:", generation_alg="uniform_rand", seed=12345) + b = [KeyGenerator.create(ks2).next_key() for _ in range(20)] + assert a == b + + +def test_uniform_rand_has_variety(): + ks = _keyspace(keys_count=1000, key_prefix="rand:", generation_alg="uniform_rand", seed=12345) + gen = KeyGenerator.create(ks) + keys = [gen.next_key() for _ in range(100)] + assert len(set(keys)) > 50 + + +def test_key_format_padding(): + # key_size_bytes=16, prefix "bench:" (6 chars) -> padding width 10. + ks = KeyspaceConfig(keys_count=100, key_size_bytes=16, key_prefix="bench:") + gen = KeyGenerator.create(ks) + key = gen.next_key() + assert key == "bench:0000000000" + + +def test_key_format_min_padding_width(): + # When prefix is longer than key_size_bytes, padding width is at least 1. + ks = KeyspaceConfig(keys_count=100, key_size_bytes=2, key_prefix="longprefix:") + gen = KeyGenerator.create(ks) + assert gen.next_key() == "longprefix:0" diff --git a/python/tests/unit/test_ndjson_writer.py b/python/tests/unit/test_ndjson_writer.py new file mode 100644 index 0000000..79354ca --- /dev/null +++ b/python/tests/unit/test_ndjson_writer.py @@ -0,0 +1,66 @@ +import json + +from resp_bench.command.command import CommandResult +from resp_bench.metrics.collector import MetricsCollector +from resp_bench.metrics.ndjson_writer import NdjsonWriter + + +def _collector_with_data(): + collector = MetricsCollector() + collector.start() + for latency in (100, 200, 300, 400, 500): + collector.record(CommandResult(command_name="GET", latency_micros=latency, success=True)) + collector.record(CommandResult(command_name="SET", latency_micros=1000, success=True)) + collector.record(CommandResult(command_name="SET", latency_micros=0, success=False)) + collector.stop() + return collector + + +def test_ndjson_schema(tmp_path): + out = tmp_path / "metrics.ndjson" + writer = NdjsonWriter(str(out)) + writer.set_metadata( + commit_id="abc123", + driver_id="redis-py", + primary_driver_version="8.1.0", + ) + writer.write_phase_results( + phase_id="STEADY", + status="COMPLETED", + connections=4, + collector=_collector_with_data(), + ) + + lines = out.read_text().splitlines() + assert len(lines) == 1 + obj = json.loads(lines[0]) + + assert obj["metadata"]["driver_id"] == "redis-py" + assert obj["metadata"]["commit_id"] == "abc123" + assert obj["metadata"]["timestamp"].endswith("Z") + + assert obj["phase"]["id"] == "STEADY" + assert obj["phase"]["status"] == "COMPLETED" + assert obj["phase"]["connections"] == 4 + assert obj["phase"]["start_timestamp"].endswith("Z") + + assert obj["totals"]["requests"] == 7 + assert obj["totals"]["errors"] == 1 + + # Command keys are uppercased. + assert set(obj["metrics"].keys()) == {"GET", "SET"} + get = obj["metrics"]["GET"] + assert get["requests"] == 5 + assert get["errors"] == 0 + assert get["latency"]["unit"] == "us" + assert get["latency"]["count"] == 5 + summary = get["latency"]["summary"] + assert set(summary.keys()) == {"min", "p50", "p95", "p99", "p999", "max"} + assert all(isinstance(v, int) for v in summary.values()) + assert get["latency"]["hdr"]["format"] == "hdr" + assert get["latency"]["hdr"]["sigfig"] == 3 + assert get["latency"]["hdr"]["payload_b64"].startswith("HIST") + + # SET had 1 success + 1 error. + assert obj["metrics"]["SET"]["requests"] == 2 + assert obj["metrics"]["SET"]["errors"] == 1 diff --git a/python/tests/unit/test_rate_limiter.py b/python/tests/unit/test_rate_limiter.py new file mode 100644 index 0000000..72080c3 --- /dev/null +++ b/python/tests/unit/test_rate_limiter.py @@ -0,0 +1,25 @@ +import time + +from resp_bench.engine.rate_limiter import RateLimiter + + +def test_unlimited_when_rate_non_positive(): + assert RateLimiter.create(0) is None + assert RateLimiter.create(-1) is None + + +async def test_enforces_rate_without_exceeding(): + rate = 500 # ops/sec -> 2ms interval + n = 50 + limiter = RateLimiter.create(rate) + start = time.monotonic() + for _ in range(n): + await limiter.acquire() + elapsed = time.monotonic() - start + + # Leaky bucket: first acquire is immediate, so ~ (n-1) intervals expected. + expected = (n - 1) / rate + # The limiter must not let us exceed the target rate (allow 10% slack for + # sleep granularity); the upper bound is loose to avoid CI flakiness. + assert elapsed >= expected * 0.9 + assert elapsed <= expected * 3.0 diff --git a/scripts/generate_graphs.py b/scripts/generate_graphs.py index 177a7e3..6962a57 100644 --- a/scripts/generate_graphs.py +++ b/scripts/generate_graphs.py @@ -71,8 +71,9 @@ # C# drivers "stackexchange-redis": "csharp", "valkey-glide-csharp": "csharp", - # Python drivers (future) + # Python drivers "redis-py": "python", + "valkey-glide-python": "python", "aioredis": "python", } diff --git a/scripts/run_benchmark_matrix.py b/scripts/run_benchmark_matrix.py index 4bfcf9d..1c0faff 100644 --- a/scripts/run_benchmark_matrix.py +++ b/scripts/run_benchmark_matrix.py @@ -124,6 +124,9 @@ # C# drivers "stackexchange-redis": "csharp", "valkey-glide-csharp": "csharp", + # Python drivers + "redis-py": "python", + "valkey-glide-python": "python", # Recording (default to java) "recording": "java", }