Function benchmark_arbiter_forwarding

benchmarks/dirty_streaming.py:248–282

Benchmark arbiter message forwarding throughput.

(results, num_chunks=1000)

async def benchmark_arbiter_forwarding(results, num_chunks=1000):
    """Benchmark arbiter message forwarding throughput."""
    arbiter = create_arbiter()

    messages = []
    for i in range(num_chunks):
        messages.append(make_chunk_message(f"bench-{i}", f"data-{i}"))
    messages.append(make_end_message(f"bench-{num_chunks}"))

    mock_reader = MockStreamReader(messages)

    async def mock_get_connection(pid):
        return mock_reader, MockStreamWriter()

    arbiter._get_worker_connection = mock_get_connection

    client_writer = MockStreamWriter()

    gc.collect()
    start = time.perf_counter()

    request = make_request("bench-forward", "benchmark:App", "stream")
    await arbiter._execute_on_worker(1234, request, client_writer)

    duration = time.perf_counter() - start

    results.add(
        f"Arbiter forwarding ({num_chunks} chunks)",
        iterations=1,
        duration=duration,
        chunks=num_chunks,
        bytes_total=client_writer.bytes_written
    )

    arbiter._cleanup_sync()


async def benchmark_streaming_latency(results, iterations=100):