Method run_benchmark

benchmarks/dirty_benchmark.py:284–367

Run a benchmark with specified parameters. Each concurrent worker maintains a persistent connection to the arbiter and makes sequential requests. This simulates how real HTTP workers use the dirty client (one connection per worker thread). Args:

(
        self,
        action: str,
        args: tuple = (),
        kwargs: dict = None,
        total_requests: int = 1000,
        concurrency: int = 10,
        timeout: float = 30.0,
    )

                client.execute(BENCHMARK_APP, "health")

    def run_benchmark(
        self,
        action: str,
        args: tuple = (),
        kwargs: dict = None,
        total_requests: int = 1000,
        concurrency: int = 10,
        timeout: float = 30.0,
    ) -> tuple[list[float], list[str]]:
        """
        Run a benchmark with specified parameters.

        Each concurrent worker maintains a persistent connection to the arbiter
        and makes sequential requests. This simulates how real HTTP workers
        use the dirty client (one connection per worker thread).

        Args:
            action: Action to call on the benchmark app
            args: Positional arguments for the action
            kwargs: Keyword arguments for the action
            total_requests: Total number of requests to make
            concurrency: Number of concurrent clients
            timeout: Timeout per request in seconds

        Returns:
            Tuple of (latencies in ms, error messages)
        """
        kwargs = kwargs or {}
        latencies = []
        errors = []
        lock = threading.Lock()

        # Calculate requests per worker
        requests_per_worker = total_requests // concurrency
        remainder = total_requests % concurrency

        def worker_task(num_requests: int) -> None:
            """Worker that makes sequential requests on a persistent connection."""
            worker_latencies = []
            worker_errors = []

            try:
                client = DirtyClient(self.socket_path, timeout=timeout)
                client.connect()

                for _ in range(num_requests):
                    try:
                        start = time.perf_counter()
                        client.execute(BENCHMARK_APP, action, *args, **kwargs)
                        elapsed = (time.perf_counter() - start) * 1000
                        worker_latencies.append(elapsed)
                    except Exception as e:
                        worker_errors.append(str(e))
                        # Reconnect on error
                        try:
                            client.close()
                            client = DirtyClient(self.socket_path, timeout=timeout)
                            client.connect()