Function run_stream

benchmarks/dirty_streaming.py:334–349

(stream_id)

    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()