Class BenchmarkResults

benchmarks/dirty_streaming.py:161–204

Store and display benchmark results.

class BenchmarkResults:
    """Store and display benchmark results."""

    def __init__(self):
        self.results = []

    def add(self, name, iterations, duration, chunks=None, bytes_total=None,
            memory_start=None, memory_end=None):
        throughput = iterations / duration if duration > 0 else 0
        result = {
            "name": name,
            "iterations": iterations,
            "duration_s": round(duration, 4),
            "throughput_per_s": round(throughput, 2),
        }
        if chunks:
            result["chunks_per_s"] = round(chunks / duration, 2)
        if bytes_total:
            result["mb_per_s"] = round(bytes_total / (1024 * 1024) / duration, 2)
        if memory_start is not None and memory_end is not None:
            result["memory_start_mb"] = round(memory_start / (1024 * 1024), 2)
            result["memory_end_mb"] = round(memory_end / (1024 * 1024), 2)
            result["memory_delta_mb"] = round((memory_end - memory_start) / (1024 * 1024), 2)
        self.results.append(result)

    def display(self):
        print("\n" + "=" * 70)
        print("BENCHMARK RESULTS")
        print("=" * 70)
        for result in self.results:
            print(f"\n{result['name']}")
            print("-" * 50)
            for key, value in result.items():
                if key != "name":
                    print(f"  {key}: {value}")
        print("\n" + "=" * 70)

    def save_json(self, filepath):
        with open(filepath, 'w') as f:
            json.dump({
                "timestamp": datetime.now().isoformat(),
                "results": self.results
            }, f, indent=2)
        print(f"Results saved to {filepath}")


async def benchmark_worker_streaming_throughput(results, chunk_size=1024, num_chunks=1000):