From 31918991199fdea18f165c0d000527e51547b550 Mon Sep 17 00:00:00 2001 From: James Duong Date: Mon, 6 Apr 2026 12:28:39 -0700 Subject: [PATCH] Add Python benchmark engine with sync and async runners Implement the Python engine for resp-bench with support for both synchronous (threaded) and asynchronous (asyncio) execution modes. Supported drivers: - redis-py (sync + async) - valkey-py (sync + async) - valkey-glide (sync + async) Features: - HdrHistogram latency collection (1us-600s, 3 significant figures) - NDJSON metrics output compatible with graph generator - Java-compatible deterministic key generation (LCG PRNG) - Token bucket rate limiting (CPS/RPS) - Request-count and duration-based completion criteria - Weighted command selection (get, set, ping) Integration: - Makefile targets: python-build/run/run-sync/run-async/test/clean/info - Driver configs: default, high-throughput, and example for all 3 drivers - Matrix runner: added redis-py, valkey-py, valkey-glide-python to DRIVER_ENGINE_MAP Signed-off-by: James Duong --- Makefile | 48 +++- configs/drivers/default/redis-py.json | 7 + .../drivers/default/valkey-glide-python.json | 7 + configs/drivers/default/valkey-py.json | 7 + .../drivers/example-redis-py-standalone.json | 7 + ...xample-valkey-glide-python-standalone.json | 7 + .../drivers/example-valkey-py-standalone.json | 7 + configs/drivers/high-throughput/redis-py.json | 8 + .../high-throughput/valkey-glide-python.json | 8 + .../drivers/high-throughput/valkey-py.json | 8 + python/README.md | 117 ++++---- python/pyproject.toml | 25 ++ python/requirements.txt | 3 + python/src/resp_bench/__init__.py | 1 + python/src/resp_bench/__main__.py | 73 +++++ python/src/resp_bench/client/__init__.py | 1 + .../src/resp_bench/client/benchmark_client.py | 84 ++++++ .../client/benchmark_client_factory.py | 62 ++++ python/src/resp_bench/client/impl/__init__.py | 1 + .../resp_bench/client/impl/redis_py_client.py | 78 ++++++ .../client/impl/valkey_glide_client.py | 69 +++++ .../client/impl/valkey_py_client.py | 78 ++++++ python/src/resp_bench/client/timed_result.py | 15 + python/src/resp_bench/command/__init__.py | 1 + python/src/resp_bench/command/command.py | 26 ++ .../src/resp_bench/command/command_factory.py | 25 ++ .../src/resp_bench/command/impl/__init__.py | 1 + .../resp_bench/command/impl/get_command.py | 23 ++ .../resp_bench/command/impl/ping_command.py | 21 ++ .../resp_bench/command/impl/set_command.py | 30 ++ python/src/resp_bench/config/__init__.py | 1 + python/src/resp_bench/config/config_loader.py | 84 ++++++ python/src/resp_bench/config/models.py | 76 +++++ python/src/resp_bench/engine/__init__.py | 1 + .../src/resp_bench/engine/benchmark_engine.py | 264 ++++++++++++++++++ .../src/resp_bench/engine/command_selector.py | 23 ++ python/src/resp_bench/engine/java_random.py | 29 ++ python/src/resp_bench/engine/key_generator.py | 37 +++ python/src/resp_bench/engine/rate_limiter.py | 49 ++++ python/src/resp_bench/metrics/__init__.py | 1 + .../resp_bench/metrics/metrics_collector.py | 102 +++++++ .../src/resp_bench/metrics/ndjson_writer.py | 111 ++++++++ scripts/run_benchmark_matrix.py | 4 + 43 files changed, 1568 insertions(+), 62 deletions(-) create mode 100644 configs/drivers/default/redis-py.json create mode 100644 configs/drivers/default/valkey-glide-python.json create mode 100644 configs/drivers/default/valkey-py.json create mode 100644 configs/drivers/example-redis-py-standalone.json create mode 100644 configs/drivers/example-valkey-glide-python-standalone.json create mode 100644 configs/drivers/example-valkey-py-standalone.json create mode 100644 configs/drivers/high-throughput/redis-py.json create mode 100644 configs/drivers/high-throughput/valkey-glide-python.json create mode 100644 configs/drivers/high-throughput/valkey-py.json create mode 100644 python/pyproject.toml create mode 100644 python/requirements.txt create mode 100644 python/src/resp_bench/__init__.py create mode 100644 python/src/resp_bench/__main__.py create mode 100644 python/src/resp_bench/client/__init__.py create mode 100644 python/src/resp_bench/client/benchmark_client.py create mode 100644 python/src/resp_bench/client/benchmark_client_factory.py create mode 100644 python/src/resp_bench/client/impl/__init__.py create mode 100644 python/src/resp_bench/client/impl/redis_py_client.py create mode 100644 python/src/resp_bench/client/impl/valkey_glide_client.py create mode 100644 python/src/resp_bench/client/impl/valkey_py_client.py create mode 100644 python/src/resp_bench/client/timed_result.py create mode 100644 python/src/resp_bench/command/__init__.py create mode 100644 python/src/resp_bench/command/command.py create mode 100644 python/src/resp_bench/command/command_factory.py create mode 100644 python/src/resp_bench/command/impl/__init__.py create mode 100644 python/src/resp_bench/command/impl/get_command.py create mode 100644 python/src/resp_bench/command/impl/ping_command.py create mode 100644 python/src/resp_bench/command/impl/set_command.py create mode 100644 python/src/resp_bench/config/__init__.py create mode 100644 python/src/resp_bench/config/config_loader.py create mode 100644 python/src/resp_bench/config/models.py create mode 100644 python/src/resp_bench/engine/__init__.py create mode 100644 python/src/resp_bench/engine/benchmark_engine.py create mode 100644 python/src/resp_bench/engine/command_selector.py create mode 100644 python/src/resp_bench/engine/java_random.py create mode 100644 python/src/resp_bench/engine/key_generator.py create mode 100644 python/src/resp_bench/engine/rate_limiter.py create mode 100644 python/src/resp_bench/metrics/__init__.py create mode 100644 python/src/resp_bench/metrics/metrics_collector.py create mode 100644 python/src/resp_bench/metrics/ndjson_writer.py diff --git a/Makefile b/Makefile index eb1e22e..090cea8 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-run-async python-run-sync 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" @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-info Show supported Python drivers and commands" @echo "" @echo "Ruby Engine:" @echo " make ruby-build Install Ruby dependencies" @@ -305,24 +306,43 @@ 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 + cd python && python -m resp_bench \ + --server $(SERVER) \ + --driver ../$(DRIVER) \ + --workload ../$(WORKLOAD) \ + --metrics ../$(METRICS_OUTPUT) + +python-run-async: python-build + cd python && python -m resp_bench \ + --server $(SERVER) \ + --driver ../$(DRIVER) \ + --workload ../$(WORKLOAD) \ + --metrics ../$(METRICS_OUTPUT) \ + --mode async + +python-run-sync: python-build + cd python && python -m resp_bench \ + --server $(SERVER) \ + --driver ../$(DRIVER) \ + --workload ../$(WORKLOAD) \ + --metrics ../$(METRICS_OUTPUT) \ + --mode sync python-clean: - @echo "Python engine not yet implemented" - @echo "Placeholder for: cd python && rm -rf __pycache__ *.egg-info dist build" + cd python && rm -rf __pycache__ *.egg-info dist build src/*.egg-info + +python-info: python-build + cd python && python -m resp_bench --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..48b67f6 --- /dev/null +++ b/configs/drivers/default/redis-py.json @@ -0,0 +1,7 @@ +{ + "schema_version": "1.0", + "description": "redis-py client - default configuration", + "driver_id": "redis-py", + "mode": "standalone", + "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..63b5c84 --- /dev/null +++ b/configs/drivers/default/valkey-glide-python.json @@ -0,0 +1,7 @@ +{ + "schema_version": "1.0", + "description": "valkey-glide Python client - default configuration", + "driver_id": "valkey-glide-python", + "mode": "standalone", + "specific_driver_config": {} +} diff --git a/configs/drivers/default/valkey-py.json b/configs/drivers/default/valkey-py.json new file mode 100644 index 0000000..7507365 --- /dev/null +++ b/configs/drivers/default/valkey-py.json @@ -0,0 +1,7 @@ +{ + "schema_version": "1.0", + "description": "valkey-py client - default configuration", + "driver_id": "valkey-py", + "mode": "standalone", + "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..3643256 --- /dev/null +++ b/configs/drivers/example-redis-py-standalone.json @@ -0,0 +1,7 @@ +{ + "schema_version": "1.0", + "description": "redis-py client - standalone mode", + "driver_id": "redis-py", + "mode": "standalone", + "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..37bc542 --- /dev/null +++ b/configs/drivers/example-valkey-glide-python-standalone.json @@ -0,0 +1,7 @@ +{ + "schema_version": "1.0", + "description": "valkey-glide Python client - standalone mode", + "driver_id": "valkey-glide-python", + "mode": "standalone", + "specific_driver_config": {} +} diff --git a/configs/drivers/example-valkey-py-standalone.json b/configs/drivers/example-valkey-py-standalone.json new file mode 100644 index 0000000..5a43fd1 --- /dev/null +++ b/configs/drivers/example-valkey-py-standalone.json @@ -0,0 +1,7 @@ +{ + "schema_version": "1.0", + "description": "valkey-py client - standalone mode", + "driver_id": "valkey-py", + "mode": "standalone", + "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..65788f2 --- /dev/null +++ b/configs/drivers/high-throughput/redis-py.json @@ -0,0 +1,8 @@ +{ + "schema_version": "1.0", + "description": "redis-py 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..13c37bd --- /dev/null +++ b/configs/drivers/high-throughput/valkey-glide-python.json @@ -0,0 +1,8 @@ +{ + "schema_version": "1.0", + "description": "valkey-glide Python client - high-throughput configuration", + "driver_id": "valkey-glide-python", + "mode": "standalone", + "specific_driver_config": {}, + "command_timeout_ms": 10000 +} diff --git a/configs/drivers/high-throughput/valkey-py.json b/configs/drivers/high-throughput/valkey-py.json new file mode 100644 index 0000000..f1b4edb --- /dev/null +++ b/configs/drivers/high-throughput/valkey-py.json @@ -0,0 +1,8 @@ +{ + "schema_version": "1.0", + "description": "valkey-py client - high-throughput configuration", + "driver_id": "valkey-py", + "mode": "standalone", + "specific_driver_config": {}, + "command_timeout_ms": 10000 +} diff --git a/python/README.md b/python/README.md index b009bac..43c586d 100644 --- a/python/README.md +++ b/python/README.md @@ -1,67 +1,88 @@ # resp-bench Python Engine -๐Ÿšง **This engine is planned but not yet implemented.** +Python implementation of the resp-bench benchmark suite, supporting both sync and async execution modes. -## Overview +## Supported Drivers -Python implementation of the resp-bench benchmark suite. +| Driver | Package | Sync | Async | +|--------|---------|------|-------| +| redis-py | `redis` | โœ… | โœ… | +| valkey-py | `valkey` | โœ… | โœ… | +| valkey-glide | `valkey-glide` | โœ… | โœ… | -## Planned Drivers +## Installation -| Driver | Package | Status | -|--------|---------|--------| -| redis-py | `redis` | ๐Ÿ“‹ Planned | -| redis-py-async | `redis[hiredis]` | ๐Ÿ“‹ Planned | -| valkey-glide | `valkey-glide` | ๐Ÿ“‹ Planned | - -## Planned Features - -- Full parity with Java engine -- Async/await based execution using `asyncio` -- HdrHistogram for latency collection -- NDJSON metrics output - -## Contributing - -We welcome contributions to implement the Python engine! Please see: -- [Architecture Documentation](../docs/ARCHITECTURE.md) -- [Adding a Language Guide](../docs/ADDING_LANGUAGE.md) - -## Directory Structure (Planned) - -``` -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/ +```bash +cd python +pip install -e . ``` -## Usage (Future) +## Usage ```bash -# Install -pip install -e . - -# Run benchmark +# Run benchmark (async mode, default) python -m resp_bench \ --server localhost:6379 \ --driver ../configs/drivers/example-redis-py-standalone.json \ --workload ../configs/workloads/example-workload.json \ --metrics output.ndjson +# Run in sync mode +python -m resp_bench \ + --server localhost:6379 \ + --driver ../configs/drivers/example-redis-py-standalone.json \ + --workload ../configs/workloads/example-workload.json \ + --metrics output.ndjson \ + --mode sync + # Show supported drivers python -m resp_bench --info ``` + +## Using with Makefile + +```bash +# Default (async) +make python-run DRIVER=configs/drivers/example-redis-py-standalone.json \ + WORKLOAD=configs/workloads/example-workload.json + +# Explicit sync +make python-run-sync DRIVER=configs/drivers/example-redis-py-standalone.json \ + WORKLOAD=configs/workloads/example-workload.json + +# Explicit async +make python-run-async DRIVER=configs/drivers/example-redis-py-standalone.json \ + WORKLOAD=configs/workloads/example-workload.json +``` + +## Testing valkey-glide PRs + +To benchmark a specific valkey-glide PR branch: + +```bash +# Install the PR branch +pip install git+https://github.com/valkey-io/valkey-glide.git@#subdirectory=python + +# Run benchmark +make python-run \ + DRIVER=configs/drivers/example-valkey-glide-python-standalone.json \ + WORKLOAD=configs/workloads/example-workload.json +``` + +## Architecture + +- **Async mode**: Uses `asyncio` with one coroutine per connection. Best for I/O-bound workloads with many connections. +- **Sync mode**: Uses `ThreadPoolExecutor` with one thread per connection. Each thread has its own `MetricsCollector` to avoid lock contention, merged after the phase completes. + +Both modes support: +- HdrHistogram latency collection (1ยตsโ€“600s, 3 significant figures) +- NDJSON output compatible with the resp-bench graph generator +- Java-compatible deterministic key generation (LCG PRNG) +- Token bucket rate limiting (CPS/RPS) +- Request-count and duration-based completion criteria + +## Supported Commands + +- `get` โ€” GET key +- `set` โ€” SET key value +- `ping` โ€” PING diff --git a/python/pyproject.toml b/python/pyproject.toml new file mode 100644 index 0000000..eb944a5 --- /dev/null +++ b/python/pyproject.toml @@ -0,0 +1,25 @@ +[build-system] +requires = ["setuptools>=68.0"] +build-backend = "setuptools.build_meta" + +[project] +name = "resp-bench" +version = "1.0.0" +description = "Python implementation of resp-bench benchmark suite" +requires-python = ">=3.9" +license = "Apache-2.0" +dependencies = [ + "redis>=5.0", + "valkey-glide>=1.0", + "hdrhistogram>=0.10", +] + +[project.optional-dependencies] +async = ["redis>=5.0", "valkey-glide>=1.0"] +dev = ["pytest>=7.0"] + +[project.scripts] +resp-bench = "resp_bench.__main__:main" + +[tool.setuptools.packages.find] +where = ["src"] diff --git a/python/requirements.txt b/python/requirements.txt new file mode 100644 index 0000000..7c63ab8 --- /dev/null +++ b/python/requirements.txt @@ -0,0 +1,3 @@ +redis>=5.0 +valkey-glide>=1.0 +hdrhistogram>=0.10 diff --git a/python/src/resp_bench/__init__.py b/python/src/resp_bench/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/python/src/resp_bench/__init__.py @@ -0,0 +1 @@ + diff --git a/python/src/resp_bench/__main__.py b/python/src/resp_bench/__main__.py new file mode 100644 index 0000000..d61875c --- /dev/null +++ b/python/src/resp_bench/__main__.py @@ -0,0 +1,73 @@ +"""resp-bench Python engine CLI entry point.""" + +import argparse +import asyncio +import json +import sys + +from .client.benchmark_client_factory import BenchmarkClientFactory +from .config.config_loader import ConfigLoader +from .engine.benchmark_engine import BenchmarkEngine + + +def print_info(): + drivers = BenchmarkClientFactory.supported_drivers() + print("resp-bench Python Engine") + print() + print("Supported drivers:") + for d in drivers: + print(f" - {d}") + print() + print("Supported commands: get, set, ping") + + +def main(): + parser = argparse.ArgumentParser(description="resp-bench Python engine") + parser.add_argument("--server", help="Server address (host:port)") + parser.add_argument("--driver", help="Driver config JSON file") + parser.add_argument("--workload", help="Workload config JSON file") + parser.add_argument("--metrics", help="Metrics output file (NDJSON)") + parser.add_argument("--commit-id", help="Git commit ID for metadata") + parser.add_argument( + "--mode", + choices=["sync", "async"], + default="async", + help="Execution mode (default: async)", + ) + parser.add_argument( + "--info", action="store_true", help="Show supported drivers" + ) + + args = parser.parse_args() + + if args.info: + print_info() + return + + if not all([args.server, args.driver, args.workload, args.metrics]): + parser.error("--server, --driver, --workload, and --metrics are required") + + driver_config = ConfigLoader.load_driver_config(args.driver) + workload_config = ConfigLoader.load_workload_config(args.workload) + + host, _, port_str = args.server.partition(":") + port = int(port_str) if port_str else 6379 + + engine = BenchmarkEngine( + host=host, + port=port, + driver_config=driver_config, + workload_config=workload_config, + metrics_path=args.metrics, + commit_id=args.commit_id, + mode=args.mode, + ) + + if args.mode == "async": + asyncio.run(engine.run_async()) + else: + engine.run_sync() + + +if __name__ == "__main__": + 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..8b13789 --- /dev/null +++ b/python/src/resp_bench/client/__init__.py @@ -0,0 +1 @@ + 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..c2a1d2c --- /dev/null +++ b/python/src/resp_bench/client/benchmark_client.py @@ -0,0 +1,84 @@ +"""Abstract benchmark client interface.""" + +import time +from abc import ABC, abstractmethod +from typing import List + +from ..config.models import DriverConfig +from .timed_result import TimedResult + + +class BenchmarkClient(ABC): + """Base class for sync benchmark clients.""" + + @abstractmethod + def connect(self, host: str, port: int, config: DriverConfig) -> None: + pass + + @abstractmethod + def ping(self) -> TimedResult: + pass + + @abstractmethod + def get(self, key: str) -> TimedResult: + pass + + @abstractmethod + def set(self, key: str, value: bytes) -> TimedResult: + pass + + @abstractmethod + def close(self) -> None: + pass + + @abstractmethod + def driver_version(self) -> str: + pass + + def _measure(self, func): + start = time.perf_counter_ns() + try: + result = func() + latency = (time.perf_counter_ns() - start) // 1000 + return TimedResult(value=result, latency_micros=latency) + except Exception as e: + latency = (time.perf_counter_ns() - start) // 1000 + return TimedResult(latency_micros=latency, error=e) + + +class AsyncBenchmarkClient(ABC): + """Base class for async benchmark clients.""" + + @abstractmethod + async def connect(self, host: str, port: int, config: DriverConfig) -> None: + pass + + @abstractmethod + async def ping(self) -> TimedResult: + pass + + @abstractmethod + async def get(self, key: str) -> TimedResult: + pass + + @abstractmethod + async def set(self, key: str, value: bytes) -> TimedResult: + pass + + @abstractmethod + async def close(self) -> None: + pass + + @abstractmethod + def driver_version(self) -> str: + pass + + async def _measure(self, coro): + start = time.perf_counter_ns() + try: + result = await coro + latency = (time.perf_counter_ns() - start) // 1000 + return TimedResult(value=result, latency_micros=latency) + except Exception as e: + latency = (time.perf_counter_ns() - start) // 1000 + return TimedResult(latency_micros=latency, error=e) diff --git a/python/src/resp_bench/client/benchmark_client_factory.py b/python/src/resp_bench/client/benchmark_client_factory.py new file mode 100644 index 0000000..9f8f02d --- /dev/null +++ b/python/src/resp_bench/client/benchmark_client_factory.py @@ -0,0 +1,62 @@ +"""Factory for creating benchmark clients by driver ID.""" + + +# Lazy imports to avoid requiring all client libraries at once +_SYNC_DRIVERS = { + "redis-py": ("resp_bench.client.impl.redis_py_client", "RedisPySyncClient"), + "valkey-py": ("resp_bench.client.impl.valkey_py_client", "ValkeyPySyncClient"), + "valkey-glide": ( + "resp_bench.client.impl.valkey_glide_client", + "ValkeyGlideSyncClient", + ), + "valkey-glide-python": ( + "resp_bench.client.impl.valkey_glide_client", + "ValkeyGlideSyncClient", + ), +} + +_ASYNC_DRIVERS = { + "redis-py": ("resp_bench.client.impl.redis_py_client", "RedisPyAsyncClient"), + "valkey-py": ("resp_bench.client.impl.valkey_py_client", "ValkeyPyAsyncClient"), + "valkey-glide": ( + "resp_bench.client.impl.valkey_glide_client", + "ValkeyGlideAsyncClient", + ), + "valkey-glide-python": ( + "resp_bench.client.impl.valkey_glide_client", + "ValkeyGlideAsyncClient", + ), +} + + +def _load_class(module_name: str, class_name: str): + import importlib + + mod = importlib.import_module(module_name) + return getattr(mod, class_name) + + +class BenchmarkClientFactory: + @staticmethod + def create_sync(driver_id: str) -> BenchmarkClient: + if driver_id not in _SYNC_DRIVERS: + raise ValueError( + f"Unknown driver: {driver_id}. " + f"Supported: {', '.join(_SYNC_DRIVERS.keys())}" + ) + mod, cls = _SYNC_DRIVERS[driver_id] + return _load_class(mod, cls)() + + @staticmethod + def create_async(driver_id: str) -> AsyncBenchmarkClient: + if driver_id not in _ASYNC_DRIVERS: + raise ValueError( + f"Unknown driver: {driver_id}. " + f"Supported: {', '.join(_ASYNC_DRIVERS.keys())}" + ) + mod, cls = _ASYNC_DRIVERS[driver_id] + return _load_class(mod, cls)() + + @staticmethod + def supported_drivers(): + return list(_SYNC_DRIVERS.keys()) 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..8b13789 --- /dev/null +++ b/python/src/resp_bench/client/impl/__init__.py @@ -0,0 +1 @@ + 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..ffb1bbc --- /dev/null +++ b/python/src/resp_bench/client/impl/redis_py_client.py @@ -0,0 +1,78 @@ +"""redis-py sync and async client implementations.""" + +import redis +import redis.asyncio as aioredis + +from ...config.models import DriverConfig +from ..benchmark_client import AsyncBenchmarkClient, BenchmarkClient +from ..timed_result import TimedResult + + +class RedisPySyncClient(BenchmarkClient): + def __init__(self): + self._client = None + + def connect(self, host: str, port: int, config: DriverConfig) -> None: + kwargs = {"host": host, "port": port, "decode_responses": False} + if config.auth: + if config.auth.get("password"): + kwargs["password"] = config.auth["password"] + if config.auth.get("username"): + kwargs["username"] = config.auth["username"] + if config.mode == "cluster": + from redis.cluster import RedisCluster + + self._client = RedisCluster(**kwargs) + else: + self._client = redis.Redis(**kwargs) + + def ping(self) -> TimedResult: + return self._measure(lambda: self._client.ping()) + + def get(self, key: str) -> TimedResult: + return self._measure(lambda: self._client.get(key)) + + def set(self, key: str, value: bytes) -> TimedResult: + return self._measure(lambda: self._client.set(key, value)) + + def close(self) -> None: + if self._client: + self._client.close() + + def driver_version(self) -> str: + return redis.__version__ + + +class RedisPyAsyncClient(AsyncBenchmarkClient): + def __init__(self): + self._client = None + + async def connect(self, host: str, port: int, config: DriverConfig) -> None: + kwargs = {"host": host, "port": port, "decode_responses": False} + if config.auth: + if config.auth.get("password"): + kwargs["password"] = config.auth["password"] + if config.auth.get("username"): + kwargs["username"] = config.auth["username"] + if config.mode == "cluster": + from redis.asyncio.cluster import RedisCluster + + self._client = RedisCluster(**kwargs) + else: + self._client = aioredis.Redis(**kwargs) + + async def ping(self) -> TimedResult: + return await self._measure(self._client.ping()) + + async def get(self, key: str) -> TimedResult: + return await self._measure(self._client.get(key)) + + async def set(self, key: str, value: bytes) -> TimedResult: + return await self._measure(self._client.set(key, value)) + + async def close(self) -> None: + if self._client: + await self._client.aclose() + + def driver_version(self) -> str: + return redis.__version__ diff --git a/python/src/resp_bench/client/impl/valkey_glide_client.py b/python/src/resp_bench/client/impl/valkey_glide_client.py new file mode 100644 index 0000000..b37f987 --- /dev/null +++ b/python/src/resp_bench/client/impl/valkey_glide_client.py @@ -0,0 +1,69 @@ +"""valkey-glide sync and async client implementations.""" + +from ...config.models import DriverConfig +from ..benchmark_client import AsyncBenchmarkClient, BenchmarkClient +from ..timed_result import TimedResult + + +class ValkeyGlideSyncClient(BenchmarkClient): + def __init__(self): + self._client = None + + def connect(self, host: str, port: int, config: DriverConfig) -> None: + from glide_sync import GlideClient, GlideClientConfiguration, NodeAddress + + addr = NodeAddress(host, port) + glide_config = GlideClientConfiguration([addr]) + self._client = GlideClient.create(glide_config) + + def ping(self) -> TimedResult: + return self._measure(lambda: self._client.ping()) + + def get(self, key: str) -> TimedResult: + return self._measure(lambda: self._client.get(key)) + + def set(self, key: str, value: bytes) -> TimedResult: + return self._measure(lambda: self._client.set(key, value)) + + def close(self) -> None: + if self._client: + self._client.close() + + def driver_version(self) -> str: + try: + import glide_sync + return getattr(glide_sync, "__version__", "unknown") + except Exception: + return "unknown" + + +class ValkeyGlideAsyncClient(AsyncBenchmarkClient): + def __init__(self): + self._client = None + + async def connect(self, host: str, port: int, config: DriverConfig) -> None: + from glide import GlideClient, GlideClientConfiguration, NodeAddress + + addr = NodeAddress(host, port) + glide_config = GlideClientConfiguration([addr]) + self._client = await GlideClient.create(glide_config) + + async def ping(self) -> TimedResult: + return await self._measure(self._client.ping()) + + async def get(self, key: str) -> TimedResult: + return await self._measure(self._client.get(key)) + + async def set(self, key: str, value: bytes) -> TimedResult: + return await self._measure(self._client.set(key, value)) + + async def close(self) -> None: + if self._client: + await self._client.close() + + def driver_version(self) -> str: + try: + import glide + return getattr(glide, "__version__", "unknown") + except Exception: + return "unknown" diff --git a/python/src/resp_bench/client/impl/valkey_py_client.py b/python/src/resp_bench/client/impl/valkey_py_client.py new file mode 100644 index 0000000..d408ac1 --- /dev/null +++ b/python/src/resp_bench/client/impl/valkey_py_client.py @@ -0,0 +1,78 @@ +"""valkey-py sync and async client implementations.""" + +import valkey +import valkey.asyncio as aiovalkey + +from ...config.models import DriverConfig +from ..benchmark_client import AsyncBenchmarkClient, BenchmarkClient +from ..timed_result import TimedResult + + +class ValkeyPySyncClient(BenchmarkClient): + def __init__(self): + self._client = None + + def connect(self, host: str, port: int, config: DriverConfig) -> None: + kwargs = {"host": host, "port": port, "decode_responses": False} + if config.auth: + if config.auth.get("password"): + kwargs["password"] = config.auth["password"] + if config.auth.get("username"): + kwargs["username"] = config.auth["username"] + if config.mode == "cluster": + from valkey.cluster import ValkeyCluster + + self._client = ValkeyCluster(**kwargs) + else: + self._client = valkey.Valkey(**kwargs) + + def ping(self) -> TimedResult: + return self._measure(lambda: self._client.ping()) + + def get(self, key: str) -> TimedResult: + return self._measure(lambda: self._client.get(key)) + + def set(self, key: str, value: bytes) -> TimedResult: + return self._measure(lambda: self._client.set(key, value)) + + def close(self) -> None: + if self._client: + self._client.close() + + def driver_version(self) -> str: + return valkey.__version__ + + +class ValkeyPyAsyncClient(AsyncBenchmarkClient): + def __init__(self): + self._client = None + + async def connect(self, host: str, port: int, config: DriverConfig) -> None: + kwargs = {"host": host, "port": port, "decode_responses": False} + if config.auth: + if config.auth.get("password"): + kwargs["password"] = config.auth["password"] + if config.auth.get("username"): + kwargs["username"] = config.auth["username"] + if config.mode == "cluster": + from valkey.asyncio.cluster import ValkeyCluster + + self._client = ValkeyCluster(**kwargs) + else: + self._client = aiovalkey.Valkey(**kwargs) + + async def ping(self) -> TimedResult: + return await self._measure(self._client.ping()) + + async def get(self, key: str) -> TimedResult: + return await self._measure(self._client.get(key)) + + async def set(self, key: str, value: bytes) -> TimedResult: + return await self._measure(self._client.set(key, value)) + + async def close(self) -> None: + if self._client: + await self._client.aclose() + + def driver_version(self) -> str: + return valkey.__version__ 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..110e193 --- /dev/null +++ b/python/src/resp_bench/client/timed_result.py @@ -0,0 +1,15 @@ +"""Timed result from a client operation.""" + +from dataclasses import dataclass +from typing import Any, Optional + + +@dataclass +class TimedResult: + value: Any = None + latency_micros: int = 0 + error: Optional[Exception] = 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..8b13789 --- /dev/null +++ b/python/src/resp_bench/command/__init__.py @@ -0,0 +1 @@ + diff --git a/python/src/resp_bench/command/command.py b/python/src/resp_bench/command/command.py new file mode 100644 index 0000000..95821cb --- /dev/null +++ b/python/src/resp_bench/command/command.py @@ -0,0 +1,26 @@ +"""Command result and base class.""" + +from dataclasses import dataclass +from typing import Optional + + +@dataclass +class CommandResult: + command_name: str + latency_micros: int + success: bool + + +class Command: + """Abstract base for benchmark commands.""" + + def __init__(self, config): + self.weight = config.weight + self.name = config.command.upper() + self.data_size_bytes = config.data_size_bytes + + def execute_sync(self, client, key_generator) -> CommandResult: + raise NotImplementedError + + async def execute_async(self, client, key_generator) -> CommandResult: + raise NotImplementedError diff --git a/python/src/resp_bench/command/command_factory.py b/python/src/resp_bench/command/command_factory.py new file mode 100644 index 0000000..80efb4d --- /dev/null +++ b/python/src/resp_bench/command/command_factory.py @@ -0,0 +1,25 @@ +"""Command factory.""" + +from .impl.get_command import GetCommand +from .impl.ping_command import PingCommand +from .impl.set_command import SetCommand + +_COMMANDS = { + "ping": PingCommand, + "get": GetCommand, + "set": SetCommand, +} + + +def create_command(config): + cls = _COMMANDS.get(config.command) + if cls is None: + raise ValueError( + f"Unknown command: {config.command}. " + f"Supported: {', '.join(_COMMANDS.keys())}" + ) + return cls(config) + + +def create_all_commands(configs): + return [create_command(c) for c in configs] 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..8b13789 --- /dev/null +++ b/python/src/resp_bench/command/impl/__init__.py @@ -0,0 +1 @@ + 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..874fd91 --- /dev/null +++ b/python/src/resp_bench/command/impl/get_command.py @@ -0,0 +1,23 @@ +"""GET command implementation.""" + +from ..command import Command, CommandResult + + +class GetCommand(Command): + def execute_sync(self, client, key_generator) -> CommandResult: + key = key_generator.next_key() + result = client.get(key) + return CommandResult( + command_name=self.name, + latency_micros=result.latency_micros, + success=result.success, + ) + + async def execute_async(self, client, key_generator) -> CommandResult: + key = key_generator.next_key() + 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..0b3b807 --- /dev/null +++ b/python/src/resp_bench/command/impl/ping_command.py @@ -0,0 +1,21 @@ +"""PING command implementation.""" + +from ..command import Command, CommandResult + + +class PingCommand(Command): + def execute_sync(self, client, key_generator) -> CommandResult: + result = client.ping() + return CommandResult( + command_name=self.name, + latency_micros=result.latency_micros, + success=result.success, + ) + + async def execute_async(self, client, key_generator) -> 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..4456d95 --- /dev/null +++ b/python/src/resp_bench/command/impl/set_command.py @@ -0,0 +1,30 @@ +"""SET command implementation.""" + +from ..command import Command, CommandResult + + +class SetCommand(Command): + def __init__(self, config): + super().__init__(config) + pattern = b"0123456789ABCDEF" + self._value = (pattern * ((self.data_size_bytes // len(pattern)) + 1))[ + : self.data_size_bytes + ] + + def execute_sync(self, client, key_generator) -> CommandResult: + key = key_generator.next_key() + result = client.set(key, self._value) + return CommandResult( + command_name=self.name, + latency_micros=result.latency_micros, + success=result.success, + ) + + async def execute_async(self, client, key_generator) -> CommandResult: + key = key_generator.next_key() + result = await client.set(key, self._value) + return CommandResult( + command_name=self.name, + latency_micros=result.latency_micros, + success=result.success, + ) diff --git a/python/src/resp_bench/config/__init__.py b/python/src/resp_bench/config/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/python/src/resp_bench/config/__init__.py @@ -0,0 +1 @@ + diff --git a/python/src/resp_bench/config/config_loader.py b/python/src/resp_bench/config/config_loader.py new file mode 100644 index 0000000..e0b1452 --- /dev/null +++ b/python/src/resp_bench/config/config_loader.py @@ -0,0 +1,84 @@ +"""Load configuration from JSON files.""" + +import json + +from .models import ( + CommandConfig, + CompletionConfig, + DriverConfig, + KeyspaceConfig, + PhaseConfig, + WorkloadConfig, +) + + +class ConfigLoader: + @staticmethod + def load_driver_config(path: str) -> DriverConfig: + with open(path) as f: + data = json.load(f) + return ConfigLoader._parse_driver_config(data) + + @staticmethod + def load_workload_config(path: str) -> WorkloadConfig: + with open(path) as f: + data = json.load(f) + return ConfigLoader._parse_workload_config(data) + + @staticmethod + def _parse_driver_config(data: dict) -> 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", {}), + ) + + @staticmethod + def _parse_workload_config(data: dict) -> 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", {}), + phases=phases, + ) + + @staticmethod + def _parse_phase(data: dict) -> PhaseConfig: + completion = data.get("completion", {}) + keyspace = data.get("keyspace", {}) + commands = [ConfigLoader._parse_command(c) for c in data.get("commands", [])] + return PhaseConfig( + id=data.get("id", ""), + description=data.get("description"), + connections=data.get("connections", 1), + cps_limit=data.get("cps_limit", -1) or -1, + rps_limit=data.get("rps_limit", -1) or -1, + pipeline_depth=data.get("pipeline_depth", 1) or 1, + warmup_requests=data.get("warmup_requests", 1) or 1, + completion=CompletionConfig( + type=completion.get("type", "requests"), + seconds=completion.get("seconds"), + requests=completion.get("requests"), + ), + keyspace=KeyspaceConfig( + keys_count=keyspace.get("keys_count", 0), + key_size_bytes=keyspace.get("key_size_bytes", 16) or 16, + key_prefix=keyspace.get("key_prefix", "bench:") or "bench:", + generation_alg=keyspace.get("generation_alg", "sequential_int"), + seed=keyspace.get("seed"), + ), + commands=commands, + ) + + @staticmethod + def _parse_command(data: dict) -> CommandConfig: + return CommandConfig( + command=data.get("command", "").lower(), + weight=float(data.get("weight", 1.0)), + data_size_bytes=data.get("data_size_bytes", 256) or 256, + ) diff --git a/python/src/resp_bench/config/models.py b/python/src/resp_bench/config/models.py new file mode 100644 index 0000000..e3c7c75 --- /dev/null +++ b/python/src/resp_bench/config/models.py @@ -0,0 +1,76 @@ +"""Configuration dataclasses for resp-bench.""" + +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + + +@dataclass +class DriverConfig: + schema_version: str = "1.0" + description: Optional[str] = None + driver_id: str = "" + mode: str = "standalone" + command_timeout_ms: Optional[int] = None + tls: Optional[Dict[str, Any]] = None + auth: Optional[Dict[str, Any]] = None + specific_driver_config: Dict[str, Any] = field(default_factory=dict) + + +@dataclass +class CompletionConfig: + type: str = "requests" + seconds: Optional[int] = None + requests: Optional[int] = None + + @property + def duration_based(self) -> bool: + return self.type == "duration" + + @property + def request_based(self) -> bool: + return self.type == "requests" + + @property + def total_requests(self) -> int: + return self.requests or 0 + + @property + def duration_seconds(self) -> int: + return self.seconds or 0 + + +@dataclass +class KeyspaceConfig: + keys_count: int = 0 + key_size_bytes: int = 16 + key_prefix: str = "bench:" + generation_alg: str = "sequential_int" + seed: Optional[int] = None + + +@dataclass +class CommandConfig: + command: str = "" + weight: float = 1.0 + data_size_bytes: int = 256 + + +@dataclass +class PhaseConfig: + id: str = "" + description: Optional[str] = None + connections: int = 1 + cps_limit: int = -1 + rps_limit: int = -1 + pipeline_depth: int = 1 + warmup_requests: int = 1 + completion: CompletionConfig = field(default_factory=CompletionConfig) + keyspace: KeyspaceConfig = field(default_factory=KeyspaceConfig) + commands: List[CommandConfig] = field(default_factory=list) + + +@dataclass +class WorkloadConfig: + schema_version: str = "1.0" + benchmark_profile: Dict[str, Any] = field(default_factory=dict) + phases: List[PhaseConfig] = field(default_factory=list) diff --git a/python/src/resp_bench/engine/__init__.py b/python/src/resp_bench/engine/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/python/src/resp_bench/engine/__init__.py @@ -0,0 +1 @@ + diff --git a/python/src/resp_bench/engine/benchmark_engine.py b/python/src/resp_bench/engine/benchmark_engine.py new file mode 100644 index 0000000..021ba91 --- /dev/null +++ b/python/src/resp_bench/engine/benchmark_engine.py @@ -0,0 +1,264 @@ +"""Benchmark engine - orchestrates phases, connections, and metrics.""" + +import asyncio +import sys +import threading +import time +from concurrent.futures import ThreadPoolExecutor +from typing import Optional + +from ..client.benchmark_client_factory import BenchmarkClientFactory +from ..command.command_factory import create_all_commands +from ..config.models import DriverConfig, PhaseConfig, WorkloadConfig +from ..engine.command_selector import CommandSelector +from ..engine.key_generator import KeyGenerator +from ..engine.rate_limiter import AsyncRateLimiter, RateLimiter +from ..metrics.metrics_collector import MetricsCollector +from ..metrics.ndjson_writer import NdjsonWriter + +PROGRESS_INTERVAL = 5 # seconds + + +class BenchmarkEngine: + def __init__( + self, + host: str, + port: int, + driver_config: DriverConfig, + workload_config: WorkloadConfig, + metrics_path: str, + commit_id: Optional[str] = None, + mode: str = "async", + ): + 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 + self._mode = mode + + # โ”€โ”€ Async execution โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + + async def run_async(self): + driver_id = self._driver_config.driver_id + print(f"[INFO] Starting async benchmark with driver: {driver_id}") + + # Get driver version from a temporary client + tmp = BenchmarkClientFactory.create_async(driver_id) + version = tmp.driver_version() + self._writer.set_metadata( + commit_id=self._commit_id, + driver_id=driver_id, + driver_version=version, + ) + + for phase in self._workload_config.phases: + await self._run_phase_async(phase) + + print("[INFO] Benchmark complete.") + + async def _run_phase_async(self, phase: PhaseConfig): + print(f"[INFO] Phase '{phase.id}': {phase.connections} connections") + + # Create clients + clients = [] + for _ in range(phase.connections): + c = BenchmarkClientFactory.create_async(self._driver_config.driver_id) + await c.connect(self._host, self._port, self._driver_config) + clients.append(c) + + commands = create_all_commands(phase.commands) + key_gen = KeyGenerator(phase.keyspace) + selector = CommandSelector(commands) + rps_limiter = AsyncRateLimiter(phase.rps_limit) + collector = MetricsCollector() + + total_requests = phase.completion.total_requests + duration_secs = phase.completion.duration_seconds + collector.start() + + # Distribute work across async tasks (one per connection) + if phase.completion.request_based: + per_conn = total_requests // phase.connections + remainder = total_requests % phase.connections + tasks = [] + for i, client in enumerate(clients): + n = per_conn + (1 if i < remainder else 0) + tasks.append( + self._async_worker_requests( + client, selector, key_gen, rps_limiter, collector, n + ) + ) + await asyncio.gather(*tasks) + else: + end_time = time.monotonic() + duration_secs + tasks = [ + self._async_worker_duration( + client, selector, key_gen, rps_limiter, collector, end_time + ) + for client in clients + ] + await asyncio.gather(*tasks) + + collector.stop() + + for c in clients: + await c.close() + + self._log_phase_summary(phase, collector) + self._writer.write_phase_results( + phase_id=phase.id, + status="COMPLETED", + connections=phase.connections, + collector=collector, + ) + + async def _async_worker_requests( + self, client, selector, key_gen, limiter, collector, count + ): + for _ in range(count): + await limiter.acquire() + cmd = selector.select() + result = await cmd.execute_async(client, key_gen) + collector.record(result.command_name, result.latency_micros, result.success) + + async def _async_worker_duration( + self, client, selector, key_gen, limiter, collector, end_time + ): + while time.monotonic() < end_time: + await limiter.acquire() + cmd = selector.select() + result = await cmd.execute_async(client, key_gen) + collector.record(result.command_name, result.latency_micros, result.success) + + # โ”€โ”€ Sync execution โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + + def run_sync(self): + driver_id = self._driver_config.driver_id + print(f"[INFO] Starting sync benchmark with driver: {driver_id}") + + tmp = BenchmarkClientFactory.create_sync(driver_id) + version = tmp.driver_version() + self._writer.set_metadata( + commit_id=self._commit_id, + driver_id=driver_id, + driver_version=version, + ) + + for phase in self._workload_config.phases: + self._run_phase_sync(phase) + + print("[INFO] Benchmark complete.") + + def _run_phase_sync(self, phase: PhaseConfig): + print(f"[INFO] Phase '{phase.id}': {phase.connections} connections") + + clients = [] + for _ in range(phase.connections): + c = BenchmarkClientFactory.create_sync(self._driver_config.driver_id) + c.connect(self._host, self._port, self._driver_config) + clients.append(c) + + commands = create_all_commands(phase.commands) + key_gen = KeyGenerator(phase.keyspace) + selector = CommandSelector(commands) + rps_limiter = RateLimiter(phase.rps_limit) + + # Each thread gets its own collector, merge after + collectors = [MetricsCollector() for _ in clients] + total_requests = phase.completion.total_requests + duration_secs = phase.completion.duration_seconds + + main_collector = MetricsCollector() + main_collector.start() + + if phase.completion.request_based: + per_conn = total_requests // phase.connections + remainder = total_requests % phase.connections + with ThreadPoolExecutor(max_workers=phase.connections) as pool: + futures = [] + for i, client in enumerate(clients): + n = per_conn + (1 if i < remainder else 0) + futures.append( + pool.submit( + self._sync_worker_requests, + client, + selector, + key_gen, + rps_limiter, + collectors[i], + n, + ) + ) + for f in futures: + f.result() + else: + end_time = time.monotonic() + duration_secs + with ThreadPoolExecutor(max_workers=phase.connections) as pool: + futures = [ + pool.submit( + self._sync_worker_duration, + client, + selector, + key_gen, + rps_limiter, + collectors[i], + end_time, + ) + for i, client in enumerate(clients) + ] + for f in futures: + f.result() + + main_collector.stop() + for c in collectors: + main_collector.merge_from(c) + + for c in clients: + c.close() + + self._log_phase_summary(phase, main_collector) + self._writer.write_phase_results( + phase_id=phase.id, + status="COMPLETED", + connections=phase.connections, + collector=main_collector, + ) + + def _sync_worker_requests( + self, client, selector, key_gen, limiter, collector, count + ): + for _ in range(count): + limiter.acquire() + cmd = selector.select() + result = cmd.execute_sync(client, key_gen) + collector.record(result.command_name, result.latency_micros, result.success) + + def _sync_worker_duration( + self, client, selector, key_gen, limiter, collector, end_time + ): + while time.monotonic() < end_time: + limiter.acquire() + cmd = selector.select() + result = cmd.execute_sync(client, key_gen) + collector.record(result.command_name, result.latency_micros, result.success) + + # โ”€โ”€ Helpers โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€โ”€ + + def _log_phase_summary(self, phase, collector): + rps = 0 + if collector.duration_millis > 0: + rps = int(collector.total_requests / (collector.duration_millis / 1000)) + print( + f"[INFO] Phase '{phase.id}' complete: " + f"{collector.total_requests} requests, " + f"{collector.total_errors} errors, " + f"{collector.duration_millis}ms, " + f"~{rps} rps" + ) + for name, m in collector.all_metrics.items(): + print( + f" {name}: p50={int(m.p50)}us p99={int(m.p99)}us " + f"p999={int(m.p999)}us max={int(m.max)}us" + ) 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..acbc8f0 --- /dev/null +++ b/python/src/resp_bench/engine/command_selector.py @@ -0,0 +1,23 @@ +"""Weighted command selector.""" + +import random + + +class CommandSelector: + def __init__(self, commands): + self._commands = commands + total = sum(c.weight for c in commands) or 1.0 + cumulative = [] + s = 0.0 + for c in commands: + s += c.weight / total + cumulative.append(s) + self._cumulative = cumulative + self._rng = random.Random() + + def select(self): + r = self._rng.random() + for i, threshold in enumerate(self._cumulative): + if r <= threshold: + return self._commands[i] + return self._commands[-1] 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..2ee6b53 --- /dev/null +++ b/python/src/resp_bench/engine/java_random.py @@ -0,0 +1,29 @@ +"""Java-compatible LCG random number generator.""" + + +class JavaRandom: + """Port of java.util.Random for cross-language deterministic sequences.""" + + _MULTIPLIER = 0x5DEECE66D + _ADDEND = 0xB + _MASK = (1 << 48) - 1 + + def __init__(self, seed: int): + self._seed = (seed ^ self._MULTIPLIER) & self._MASK + + def next_int(self, bound: int) -> int: + if bound <= 0: + raise ValueError("bound must be positive") + # Power of 2 + if (bound & -bound) == bound: + return (bound * self._next_bits(31)) >> 31 + # General case - rejection sampling + while True: + bits = self._next_bits(31) + val = bits % bound + if bits - val + (bound - 1) >= 0: + return val + + def _next_bits(self, bits: int) -> int: + self._seed = ((self._seed * self._MULTIPLIER) + self._ADDEND) & self._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..fe5bfa7 --- /dev/null +++ b/python/src/resp_bench/engine/key_generator.py @@ -0,0 +1,37 @@ +"""Key generator matching Java/Ruby implementations.""" + +import threading + +from ..config.models import KeyspaceConfig +from .java_random import JavaRandom + + +class KeyGenerator: + def __init__(self, config: KeyspaceConfig): + self._keys_count = config.keys_count + self._prefix = config.key_prefix or "bench:" + self._alg = config.generation_alg + self._counter = 0 + self._lock = threading.Lock() + if self._alg == "uniform_rand": + self._rng = JavaRandom(config.seed or 0) + + def next_key(self) -> str: + if self._alg == "sequential_int": + with self._lock: + idx = self._counter % self._keys_count + self._counter += 1 + else: + with self._lock: + idx = self._rng.next_int(self._keys_count) + return f"{self._prefix}{idx}" + + def fork(self, seed: int) -> "KeyGenerator": + """Create a new generator with a different seed (for parallel workers).""" + cfg = KeyspaceConfig( + keys_count=self._keys_count, + key_prefix=self._prefix, + generation_alg=self._alg, + seed=seed, + ) + return KeyGenerator(cfg) 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..2b31d69 --- /dev/null +++ b/python/src/resp_bench/engine/rate_limiter.py @@ -0,0 +1,49 @@ +"""Token bucket rate limiter.""" + +import asyncio +import time +import threading + + +class RateLimiter: + """Leaky bucket rate limiter for sync use.""" + + def __init__(self, rate_per_second: int): + self._rate = rate_per_second + if rate_per_second > 0: + self._interval_ns = 1_000_000_000 // rate_per_second + else: + self._interval_ns = 0 + self._next_ns = time.monotonic_ns() + self._lock = threading.Lock() + + def acquire(self) -> None: + if self._rate <= 0: + return + with self._lock: + now = time.monotonic_ns() + if now < self._next_ns: + time.sleep((self._next_ns - now) / 1_000_000_000) + self._next_ns = max(now, self._next_ns) + self._interval_ns + + +class AsyncRateLimiter: + """Leaky bucket rate limiter for async use.""" + + def __init__(self, rate_per_second: int): + self._rate = rate_per_second + if rate_per_second > 0: + self._interval_ns = 1_000_000_000 // rate_per_second + else: + self._interval_ns = 0 + self._next_ns = time.monotonic_ns() + self._lock = asyncio.Lock() + + async def acquire(self) -> None: + if self._rate <= 0: + return + async with self._lock: + now = time.monotonic_ns() + if now < self._next_ns: + await asyncio.sleep((self._next_ns - now) / 1_000_000_000) + self._next_ns = max(now, self._next_ns) + self._interval_ns diff --git a/python/src/resp_bench/metrics/__init__.py b/python/src/resp_bench/metrics/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/python/src/resp_bench/metrics/__init__.py @@ -0,0 +1 @@ + diff --git a/python/src/resp_bench/metrics/metrics_collector.py b/python/src/resp_bench/metrics/metrics_collector.py new file mode 100644 index 0000000..e0b520a --- /dev/null +++ b/python/src/resp_bench/metrics/metrics_collector.py @@ -0,0 +1,102 @@ +"""Metrics collection using HdrHistogram.""" + +import threading +import time +from typing import Dict, Optional + +from hdrh.histogram import HdrHistogram + + +MAX_LATENCY = 600_000_000 # 600 seconds in microseconds + + +class CommandMetrics: + def __init__(self, command_name: str): + self.command_name = command_name + self.requests = 0 + self.errors = 0 + self.histogram = HdrHistogram(1, MAX_LATENCY, 3) + + def record(self, latency_micros: int, success: bool): + self.requests += 1 + if success: + self.histogram.record_value(min(latency_micros, MAX_LATENCY)) + else: + self.errors += 1 + + def merge_from(self, other: "CommandMetrics"): + self.requests += other.requests + self.errors += other.errors + self.histogram.add(other.histogram) + + @property + def count(self): + return self.histogram.total_count + + @property + def min(self): + return self.histogram.get_min_value() + + @property + def max(self): + return self.histogram.get_max_value() + + @property + def p50(self): + return self.histogram.get_value_at_percentile(50) + + @property + def p95(self): + return self.histogram.get_value_at_percentile(95) + + @property + def p99(self): + return self.histogram.get_value_at_percentile(99) + + @property + def p999(self): + return self.histogram.get_value_at_percentile(99.9) + + +class MetricsCollector: + def __init__(self): + self._command_metrics: Dict[str, CommandMetrics] = {} + self._lock = threading.Lock() + self.total_requests = 0 + self.total_errors = 0 + self.start_time: Optional[float] = None + self.end_time: Optional[float] = None + + def start(self): + self.start_time = time.time() + + def stop(self): + self.end_time = time.time() + + def record(self, command_name: str, latency_micros: int, success: bool): + self.total_requests += 1 + if not success: + self.total_errors += 1 + with self._lock: + if command_name not in self._command_metrics: + self._command_metrics[command_name] = CommandMetrics(command_name) + self._command_metrics[command_name].record(latency_micros, success) + + def merge_from(self, other: "MetricsCollector"): + self.total_requests += other.total_requests + self.total_errors += other.total_errors + for name, metrics in other._command_metrics.items(): + with self._lock: + if name not in self._command_metrics: + self._command_metrics[name] = CommandMetrics(name) + self._command_metrics[name].merge_from(metrics) + + @property + def duration_millis(self) -> int: + if self.start_time and self.end_time: + return int((self.end_time - self.start_time) * 1000) + return 0 + + @property + def all_metrics(self) -> Dict[str, CommandMetrics]: + return self._command_metrics 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..2360fd8 --- /dev/null +++ b/python/src/resp_bench/metrics/ndjson_writer.py @@ -0,0 +1,111 @@ +"""NDJSON metrics writer with HDR histogram encoding.""" + +import base64 +import json +import os +import time +from datetime import datetime, timezone +from typing import Optional + +from .metrics_collector import MetricsCollector + + +class NdjsonWriter: + def __init__(self, path: str): + self._path = path + self._commit_id: Optional[str] = None + self._driver_id: Optional[str] = None + self._driver_version: Optional[str] = None + + def set_metadata( + self, + commit_id: Optional[str] = None, + driver_id: Optional[str] = None, + driver_version: Optional[str] = None, + ): + self._commit_id = commit_id + self._driver_id = driver_id + self._driver_version = driver_version + + def write_phase_results( + self, + phase_id: str, + status: str, + connections: int, + collector: MetricsCollector, + ): + os.makedirs(os.path.dirname(self._path) or ".", exist_ok=True) + record = self._build_phase_json(phase_id, status, connections, collector) + with open(self._path, "a") as f: + f.write(json.dumps(record) + "\n") + + def _build_phase_json(self, phase_id, status, connections, collector): + result = {} + + if self._commit_id or self._driver_id: + metadata = {} + if self._commit_id: + metadata["commit_id"] = self._commit_id + metadata["timestamp"] = datetime.now(timezone.utc).isoformat() + if self._driver_id: + metadata["driver_id"] = self._driver_id + if self._driver_version: + metadata["primary_driver_version"] = self._driver_version + result["metadata"] = metadata + + start_ts = ( + datetime.fromtimestamp(collector.start_time, tz=timezone.utc).isoformat() + if collector.start_time + else None + ) + end_ts = ( + datetime.fromtimestamp(collector.end_time, tz=timezone.utc).isoformat() + if collector.end_time + else None + ) + + result["phase"] = { + "id": phase_id, + "status": status, + "start_timestamp": start_ts, + "finish_timestamp": end_ts, + "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): + metrics = {} + for cmd_name, cmd_metrics in collector.all_metrics.items(): + cmd_data = { + "requests": cmd_metrics.requests, + "errors": cmd_metrics.errors, + "latency": { + "unit": "us", + "count": cmd_metrics.count, + "summary": { + "min": int(cmd_metrics.min), + "p50": int(cmd_metrics.p50), + "p95": int(cmd_metrics.p95), + "p99": int(cmd_metrics.p99), + "p999": int(cmd_metrics.p999), + "max": int(cmd_metrics.max), + }, + }, + } + try: + encoded = cmd_metrics.histogram.encode() + cmd_data["latency"]["hdr"] = { + "format": "hdr", + "sigfig": 3, + "payload_b64": base64.b64encode(encoded).decode("ascii"), + } + except Exception: + pass + metrics[cmd_name] = cmd_data + return metrics diff --git a/scripts/run_benchmark_matrix.py b/scripts/run_benchmark_matrix.py index 62c8a2c..a34ca5a 100644 --- a/scripts/run_benchmark_matrix.py +++ b/scripts/run_benchmark_matrix.py @@ -87,6 +87,10 @@ # C# drivers "stackexchange-redis": "csharp", "valkey-glide-csharp": "csharp", + # Python drivers + "redis-py": "python", + "valkey-py": "python", + "valkey-glide-python": "python", # Recording (default to java) "recording": "java", }