erlang_python

#!/usr/bin/env python3
"""Benchmark for ReactorBuffer zero-copy performance.

Compares:
1. ReactorBuffer with zero-copy access (buffer protocol)
2. Regular bytes with data copying

Run: python3 examples/bench_reactor_buffer.py
"""

import time
import socket
import statistics
import sys

sys.path.insert(0, 'priv')


def bench_buffer_operations(iterations=10000):
    """Benchmark common buffer operations."""
    import erlang

    results = {}

    # Test data sizes
    sizes = [64, 256, 1024, 4096, 16384, 65536]

    for size in sizes:
        test_data = b'X' * size

        # Create ReactorBuffer
        buf = erlang.ReactorBuffer._test_create(test_data)
        regular_bytes = bytes(test_data)

        # Benchmark: memoryview access (zero-copy)
        start = time.perf_counter()
        for _ in range(iterations):
            mv = memoryview(buf)
            _ = mv[0]
        mv_time = time.perf_counter() - start

        # Benchmark: memoryview on regular bytes
        start = time.perf_counter()
        for _ in range(iterations):
            mv = memoryview(regular_bytes)
            _ = mv[0]
        mv_bytes_time = time.perf_counter() - start

        # Benchmark: extend bytearray (uses buffer protocol)
        start = time.perf_counter()
        for _ in range(iterations):
            ba = bytearray()
            ba.extend(buf)
        extend_buf_time = time.perf_counter() - start

        # Benchmark: extend bytearray from bytes
        start = time.perf_counter()
        for _ in range(iterations):
            ba = bytearray()
            ba.extend(regular_bytes)
        extend_bytes_time = time.perf_counter() - start

        # Benchmark: slice operation
        slice_len = min(100, size)
        start = time.perf_counter()
        for _ in range(iterations):
            _ = buf[0:slice_len]
        slice_buf_time = time.perf_counter() - start

        start = time.perf_counter()
        for _ in range(iterations):
            _ = regular_bytes[0:slice_len]
        slice_bytes_time = time.perf_counter() - start

        # Benchmark: startswith
        prefix = test_data[:10]
        start = time.perf_counter()
        for _ in range(iterations):
            _ = buf.startswith(prefix)
        startswith_buf_time = time.perf_counter() - start

        start = time.perf_counter()
        for _ in range(iterations):
            _ = regular_bytes.startswith(prefix)
        startswith_bytes_time = time.perf_counter() - start

        results[size] = {
            'memoryview': {'buffer': mv_time, 'bytes': mv_bytes_time},
            'extend': {'buffer': extend_buf_time, 'bytes': extend_bytes_time},
            'slice': {'buffer': slice_buf_time, 'bytes': slice_bytes_time},
            'startswith': {'buffer': startswith_buf_time, 'bytes': startswith_bytes_time},
        }

    return results


def bench_protocol_simulation(iterations=1000):
    """Simulate protocol data_received with different payload sizes."""
    import erlang

    results = {}
    sizes = [64, 256, 1024, 4096, 16384, 65536]

    for size in sizes:
        test_data = b'GET / HTTP/1.1\r\nHost: example.com\r\n\r\n' + b'X' * (size - 40)
        test_data = test_data[:size]  # Ensure exact size

        buf = erlang.ReactorBuffer._test_create(test_data)
        regular_bytes = bytes(test_data)

        # Simulate typical protocol parsing with ReactorBuffer
        def parse_with_buffer(data):
            # Check method
            if data.startswith(b'GET'):
                method = 'GET'
            elif data.startswith(b'POST'):
                method = 'POST'
            else:
                method = 'OTHER'

            # Find header end
            pos = data.find(b'\r\n\r\n')

            # Buffer to write buffer
            write_buf = bytearray()
            write_buf.extend(data)

            return len(write_buf)

        # Benchmark with ReactorBuffer
        start = time.perf_counter()
        for _ in range(iterations):
            _ = parse_with_buffer(buf)
        buf_time = time.perf_counter() - start

        # Benchmark with regular bytes
        start = time.perf_counter()
        for _ in range(iterations):
            _ = parse_with_buffer(regular_bytes)
        bytes_time = time.perf_counter() - start

        results[size] = {
            'buffer_time': buf_time,
            'bytes_time': bytes_time,
            'ops_per_sec_buffer': iterations / buf_time,
            'ops_per_sec_bytes': iterations / bytes_time,
        }

    return results


def bench_echo_protocol(iterations=500):
    """Benchmark echo protocol pattern with socketpair."""
    import erlang.reactor as reactor

    results = {}
    sizes = [64, 256, 1024, 4096, 16384]

    class EchoProtocol(reactor.Protocol):
        def data_received(self, data):
            self.write_buffer.extend(data)
            return 'write_pending'

        def write_ready(self):
            if not self.write_buffer:
                return 'read_pending'
            written = self.write(bytes(self.write_buffer))
            del self.write_buffer[:written]
            return 'continue' if self.write_buffer else 'read_pending'

    for size in sizes:
        test_data = b'X' * size
        times = []

        for _ in range(iterations):
            s1, s2 = socket.socketpair()
            s1.setblocking(False)
            s2.setblocking(False)

            try:
                reactor.set_protocol_factory(EchoProtocol)
                reactor.init_connection(s1.fileno(), {'type': 'test'})

                s2.send(test_data)

                start = time.perf_counter()
                action = reactor.on_read_ready(s1.fileno())
                elapsed = time.perf_counter() - start
                times.append(elapsed)

                reactor.close_connection(s1.fileno())
            finally:
                s1.close()
                s2.close()

        avg_time = statistics.mean(times)
        results[size] = {
            'avg_time_ms': avg_time * 1000,
            'ops_per_sec': 1.0 / avg_time,
            'p50_ms': statistics.median(times) * 1000,
            'p95_ms': sorted(times)[int(len(times) * 0.95)] * 1000,
        }

    return results


def format_results(name, results):
    """Format benchmark results."""
    print(f"\n{'='*60}")
    print(f" {name}")
    print(f"{'='*60}")

    if 'memoryview' in list(results.values())[0]:
        # Buffer operations format
        print(f"{'Size':>8} | {'Operation':>12} | {'Buffer (ms)':>12} | {'Bytes (ms)':>12} | {'Ratio':>8}")
        print("-" * 60)
        for size, ops in sorted(results.items()):
            for op_name, times in ops.items():
                buf_ms = times['buffer'] * 1000
                bytes_ms = times['bytes'] * 1000
                ratio = bytes_ms / buf_ms if buf_ms > 0 else 0
                print(f"{size:>8} | {op_name:>12} | {buf_ms:>12.3f} | {bytes_ms:>12.3f} | {ratio:>7.2f}x")
    elif 'buffer_time' in list(results.values())[0]:
        # Protocol simulation format
        print(f"{'Size':>8} | {'Buffer (ops/s)':>14} | {'Bytes (ops/s)':>14} | {'Speedup':>8}")
        print("-" * 60)
        for size, data in sorted(results.items()):
            buf_ops = data['ops_per_sec_buffer']
            bytes_ops = data['ops_per_sec_bytes']
            speedup = buf_ops / bytes_ops if bytes_ops > 0 else 0
            print(f"{size:>8} | {buf_ops:>14.0f} | {bytes_ops:>14.0f} | {speedup:>7.2f}x")
    else:
        # Echo protocol format
        print(f"{'Size':>8} | {'Avg (ms)':>10} | {'P50 (ms)':>10} | {'P95 (ms)':>10} | {'Ops/sec':>10}")
        print("-" * 60)
        for size, data in sorted(results.items()):
            print(f"{size:>8} | {data['avg_time_ms']:>10.3f} | {data['p50_ms']:>10.3f} | {data['p95_ms']:>10.3f} | {data['ops_per_sec']:>10.0f}")


def main():
    print("\n" + "=" * 60)
    print(" ReactorBuffer Zero-Copy Benchmark")
    print("=" * 60)

    # Check if ReactorBuffer is available
    try:
        import erlang
        buf = erlang.ReactorBuffer._test_create(b'test')
        print(f"ReactorBuffer available: {type(buf).__name__}")
    except Exception as e:
        print(f"ERROR: ReactorBuffer not available: {e}")
        sys.exit(1)

    print("\nRunning benchmarks...")

    # Run benchmarks
    print("\n[1/3] Buffer operations benchmark...")
    buffer_ops = bench_buffer_operations(iterations=10000)
    format_results("Buffer Operations (10000 iterations)", buffer_ops)

    print("\n[2/3] Protocol simulation benchmark...")
    protocol_sim = bench_protocol_simulation(iterations=5000)
    format_results("Protocol Simulation (5000 iterations)", protocol_sim)

    print("\n[3/3] Echo protocol benchmark...")
    echo_proto = bench_echo_protocol(iterations=200)
    format_results("Echo Protocol (200 iterations)", echo_proto)

    # Summary
    print("\n" + "=" * 60)
    print(" Summary")
    print("=" * 60)

    # Calculate average speedup for large payloads (>= 1KB)
    large_payload_speedups = []
    for size, data in protocol_sim.items():
        if size >= 1024:
            speedup = data['ops_per_sec_buffer'] / data['ops_per_sec_bytes']
            large_payload_speedups.append(speedup)

    if large_payload_speedups:
        avg_speedup = statistics.mean(large_payload_speedups)
        print(f"Average speedup for payloads >= 1KB: {avg_speedup:.2f}x")
        improvement = (avg_speedup - 1.0) * 100
        print(f"Performance improvement: {improvement:+.1f}%")

    print("\nDone.")


if __name__ == '__main__':
    main()