Function benchmark_memory_stability

benchmarks/dirty_streaming.py:372–415

Check memory stability over many iterations.

(results, iterations=10, chunks=1000)

async def benchmark_memory_stability(results, iterations=10, chunks=1000):
    """Check memory stability over many iterations."""
    worker = create_worker()

    gc.collect()
    tracemalloc.start()
    memory_samples = [tracemalloc.get_traced_memory()[0]]

    for i in range(iterations):
        writer = MockStreamWriter()

        async def gen_chunks():
            for j in range(chunks):
                yield f"chunk-{j}"

        async def mock_execute(app_path, action, args, kwargs):
            return gen_chunks()

        with mock.patch.object(worker, 'execute', side_effect=mock_execute):
            request = make_request(f"bench-mem-{i}", "benchmark:App", "stream")
            await worker.handle_request(request, writer)

        gc.collect()
        memory_samples.append(tracemalloc.get_traced_memory()[0])

    tracemalloc.stop()

    memory_start = memory_samples[0]
    memory_end = memory_samples[-1]
    memory_max = max(memory_samples)

    print(f"\nMemory stability ({iterations} iterations of {chunks} chunks):")
    print(f"  Start: {memory_start / 1024 / 1024:.2f}MB")
    print(f"  End: {memory_end / 1024 / 1024:.2f}MB")
    print(f"  Max: {memory_max / 1024 / 1024:.2f}MB")
    print(f"  Delta: {(memory_end - memory_start) / 1024 / 1024:.2f}MB")

    results.add(
        f"Memory stability ({iterations} x {chunks} chunks)",
        iterations=iterations * chunks,
        duration=0.001,  # Use small non-zero value to avoid division by zero
        memory_start=memory_start,
        memory_end=memory_end
    )


class MockClientReader: