Function run_rps_tests

tests/test-dataset-streaming.py:776–811

Run RPS (Requests Per Second) tests

(rps_test_vectors, collection, batch_size=100)

        print("No valid queries completed")

def run_rps_tests(rps_test_vectors, collection, batch_size=100):
    """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=32) as executor:
        futures = []
        for i in range(0, len(rps_test_vectors), batch_size):
            batch = rps_test_vectors[i : i + batch_size]
            futures.append(executor.submit(batch_ann_search, collection, batch))

        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 batch_ann_search(collection, vectors):
    """Perform batch ANN search using the Client class"""