Function run_rps_tests

tests/test-usv.py:434–486

Run RPS (Requests Per Second) tests

(
    rps_test_vectors,
    name,
    batch_size=100,
    top_k=10,
    early_terminate_threshold=0.0,
)

def run_rps_tests(
    rps_test_vectors,
    name,
    batch_size=100,
    top_k=10,
    early_terminate_threshold=0.0,
):
    """Run RPS (Requests Per Second) tests"""
    print(f"Using {len(rps_test_vectors)} different test vectors for RPS testing")

    start_time_rps = time.time()
    results = []

    with ThreadPoolExecutor(max_workers=8) as executor:
        futures = []
        for i in range(0, len(rps_test_vectors), batch_size):
            batch = [
                format_for_server_query(vector)
                for vector in rps_test_vectors[i : i + batch_size]
            ]
            futures.append(
                executor.submit(
                    batch_ann_search,
                    name,
                    batch,
                    top_k,
                    early_terminate_threshold,
                )
            )

        for future in as_completed(futures):
            try:
                future.result()
                results.append(True)
            except Exception as e:
                print(f"Error in RPS test: {e}")
                results.append(False)

    end_time_rps = time.time()
    actual_duration = end_time_rps - start_time_rps

    successful_requests = sum(results) * batch_size
    failed_requests = (len(results) * batch_size) - successful_requests
    total_requests = len(results) * batch_size
    rps = successful_requests / actual_duration

    print("\nRPS Test Results:")
    print(f"Total Requests: {total_requests}")
    print(f"Successful Requests: {successful_requests}")
    print(f"Failed Requests: {failed_requests}")
    print(f"Test Duration: {actual_duration:.2f} seconds")
    print(f"Requests Per Second (RPS): {rps:.2f}")
    print(f"Success Rate: {(successful_requests / total_requests * 100):.2f}%")


def main():