Function benchmark_concurrent_streams

benchmarks/dirty_streaming.py:330–369

Benchmark multiple concurrent streams.

(results, num_streams=10, chunks_per_stream=100)

async def benchmark_concurrent_streams(results, num_streams=10, chunks_per_stream=100):
    """Benchmark multiple concurrent streams."""
    arbiter = create_arbiter()

    async def run_stream(stream_id):
        messages = []
        for i in range(chunks_per_stream):
            messages.append(make_chunk_message(f"stream-{stream_id}", f"chunk-{i}"))
        messages.append(make_end_message(f"stream-{stream_id}"))

        mock_reader = MockStreamReader(messages)
        async def mock_get_connection(pid):
            return mock_reader, MockStreamWriter()

        arbiter._get_worker_connection = mock_get_connection
        client_writer = MockStreamWriter()

        request = make_request(f"bench-concurrent-{stream_id}", "benchmark:App", "stream")
        await arbiter._execute_on_worker(1234, request, client_writer)
        return len(client_writer.messages)

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

    # Run streams concurrently
    tasks = [run_stream(i) for i in range(num_streams)]
    results_list = await asyncio.gather(*tasks)

    duration = time.perf_counter() - start

    total_chunks = sum(results_list)

    results.add(
        f"Concurrent streams ({num_streams} streams, {chunks_per_stream} chunks each)",
        iterations=num_streams,
        duration=duration,
        chunks=total_chunks
    )

    arbiter._cleanup_sync()


async def benchmark_memory_stability(results, iterations=10, chunks=1000):