Add PyBuffer vs Channel benchmark

Author: benoitc

examples/bench_py_buffer.erl

@@ -0,0 +1,265 @@
+#!/usr/bin/env escript
+%% -*- erlang -*-
+%%! -pa _build/default/lib/erlang_python/ebin
+
+%%% @doc Benchmark script comparing PyBuffer with Channel API.
+%%%
+%%% Run with:
+%%%   rebar3 compile && escript examples/bench_py_buffer.erl
+
+-mode(compile).
+
+main(_Args) ->
+    io:format("~n========================================~n"),
+    io:format("PyBuffer vs Channel Benchmark~n"),
+    io:format("========================================~n~n"),
+
+    %% Start the application
+    {ok, _} = application:ensure_all_started(erlang_python),
+    {ok, _} = py:start_contexts(),
+    ok = py_channel:register_callbacks(),
+
+    %% Print system info
+    io:format("System Information:~n"),
+    io:format("  Erlang/OTP: ~s~n", [erlang:system_info(otp_release)]),
+    {ok, PyVer} = py:version(),
+    io:format("  Python: ~s~n", [PyVer]),
+    io:format("~n"),
+
+    %% Run benchmarks
+    run_write_throughput_bench(),
+    run_read_throughput_bench(),
+    run_buffer_vs_channel_bench(),
+    run_streaming_bench(),
+
+    io:format("~n========================================~n"),
+    io:format("Benchmark Complete~n"),
+    io:format("========================================~n"),
+
+    halt(0).
+
+run_write_throughput_bench() ->
+    io:format("~n--- PyBuffer Write Throughput ---~n"),
+    io:format("Writes per batch: 1000~n~n"),
+
+    Sizes = [64, 256, 1024, 4096, 16384],
+
+    io:format("~8s | ~12s | ~12s~n",
+              ["Size", "Writes/sec", "MB/sec"]),
+    io:format("~s~n", [string:copies("-", 38)]),
+
+    lists:foreach(fun(Size) ->
+        {ok, Buf} = py_buffer:new(),
+        Data = binary:copy(<<0>>, Size),
+        Iterations = 1000,
+
+        Start = erlang:monotonic_time(microsecond),
+        lists:foreach(fun(_) ->
+            ok = py_buffer:write(Buf, Data)
+        end, lists:seq(1, Iterations)),
+        End = erlang:monotonic_time(microsecond),
+
+        TotalTime = (End - Start) / 1000000,
+        WriteRate = Iterations / TotalTime,
+        MBPerSec = (Iterations * Size) / TotalTime / 1048576,
+
+        io:format("~8B | ~12w | ~12.2f~n",
+                  [Size, round(WriteRate), MBPerSec]),
+
+        py_buffer:close(Buf)
+    end, Sizes),
+    ok.
+
+run_read_throughput_bench() ->
+    io:format("~n--- PyBuffer Read Throughput (Python) ---~n"),
+    io:format("Reads per batch: 1000~n~n"),
+
+    Ctx = py:context(1),
+    ok = py:exec(Ctx, <<"
+import time
+
+def bench_buffer_reads(buf, size, iterations):
+    '''Read from buffer and measure throughput.'''
+    # Read all data
+    start = time.perf_counter()
+    total_read = 0
+    while True:
+        data = buf.read(size)
+        if not data:
+            break
+        total_read += len(data)
+    elapsed = time.perf_counter() - start
+
+    if elapsed > 0:
+        reads_per_sec = (total_read / size) / elapsed
+        mb_per_sec = total_read / elapsed / 1048576
+    else:
+        reads_per_sec = 0
+        mb_per_sec = 0
+
+    return {
+        'total_bytes': total_read,
+        'reads_per_sec': reads_per_sec,
+        'mb_per_sec': mb_per_sec,
+    }
+">>),
+
+    Sizes = [64, 256, 1024, 4096, 16384],
+
+    io:format("~8s | ~12s | ~12s~n",
+              ["Size", "Reads/sec", "MB/sec"]),
+    io:format("~s~n", [string:copies("-", 38)]),
+
+    lists:foreach(fun(Size) ->
+        Iterations = 1000,
+        {ok, Buf} = py_buffer:new(Size * Iterations),
+        Data = binary:copy(<<0>>, Size),
+
+        %% Fill the buffer
+        lists:foreach(fun(_) ->
+            ok = py_buffer:write(Buf, Data)
+        end, lists:seq(1, Iterations)),
+        ok = py_buffer:close(Buf),
+
+        %% Measure Python read performance
+        {ok, Result} = py:eval(Ctx, <<"bench_buffer_reads(buf, size, iterations)">>,
+                               #{<<"buf">> => Buf, <<"size">> => Size, <<"iterations">> => Iterations}),
+
+        ReadsPerSec = maps:get(<<"reads_per_sec">>, Result),
+        MBPerSec = maps:get(<<"mb_per_sec">>, Result),
+
+        io:format("~8B | ~12w | ~12.2f~n",
+                  [Size, round(ReadsPerSec), MBPerSec])
+    end, Sizes),
+    ok.
+
+run_buffer_vs_channel_bench() ->
+    io:format("~n--- PyBuffer vs Channel Comparison ---~n"),
+    io:format("Pattern: Erlang write/send -> Python read/receive~n"),
+    io:format("Iterations: 1000~n~n"),
+
+    Ctx = py:context(1),
+    ok = py:exec(Ctx, <<"
+from erlang.channel import Channel
+
+def read_buffer_all(buf):
+    '''Read entire buffer.'''
+    return buf.read()
+
+def recv_channel(ch_ref):
+    '''Receive from channel.'''
+    ch = Channel(ch_ref)
+    return ch.try_receive()
+">>),
+
+    Sizes = [64, 256, 1024, 4096, 16384],
+
+    io:format("~8s | ~14s | ~14s | ~8s~n",
+              ["Size", "Buffer (ops/s)", "Channel (ops/s)", "Ratio"]),
+    io:format("~s~n", [string:copies("-", 52)]),
+
+    lists:foreach(fun(Size) ->
+        Data = binary:copy(<<0>>, Size),
+        Iterations = 1000,
+
+        %% Benchmark PyBuffer: Erlang write -> Python read
+        BufStart = erlang:monotonic_time(microsecond),
+        lists:foreach(fun(_) ->
+            {ok, Buf} = py_buffer:new(Size),
+            ok = py_buffer:write(Buf, Data),
+            ok = py_buffer:close(Buf),
+            {ok, _} = py:eval(Ctx, <<"read_buffer_all(buf)">>, #{<<"buf">> => Buf})
+        end, lists:seq(1, Iterations)),
+        BufEnd = erlang:monotonic_time(microsecond),
+        BufTime = (BufEnd - BufStart) / 1000000,
+        BufOpsPerSec = Iterations / BufTime,
+
+        %% Benchmark Channel: Erlang send -> Python receive
+        {ok, Ch} = py_channel:new(),
+        ChanStart = erlang:monotonic_time(microsecond),
+        lists:foreach(fun(_) ->
+            ok = py_channel:send(Ch, Data),
+            {ok, _} = py:eval(Ctx, <<"recv_channel(ch)">>, #{<<"ch">> => Ch})
+        end, lists:seq(1, Iterations)),
+        ChanEnd = erlang:monotonic_time(microsecond),
+        ChanTime = (ChanEnd - ChanStart) / 1000000,
+        ChanOpsPerSec = Iterations / ChanTime,
+        py_channel:close(Ch),
+
+        Ratio = BufOpsPerSec / ChanOpsPerSec,
+        io:format("~8B | ~14w | ~14w | ~.2fx~n",
+                  [Size, round(BufOpsPerSec), round(ChanOpsPerSec), Ratio])
+    end, Sizes),
+    ok.
+
+run_streaming_bench() ->
+    io:format("~n--- Streaming Comparison (chunked transfer) ---~n"),
+    io:format("Total data: 1MB, varying chunk sizes~n~n"),
+
+    Ctx = py:context(1),
+    ok = py:exec(Ctx, <<"
+from erlang.channel import Channel
+
+def stream_read_buffer(buf):
+    '''Stream read entire buffer.'''
+    total = 0
+    while True:
+        chunk = buf.read(8192)  # 8KB reads
+        if not chunk:
+            break
+        total += len(chunk)
+    return total
+
+def stream_recv_channel(ch_ref, num_chunks):
+    '''Stream receive from channel.'''
+    ch = Channel(ch_ref)
+    total = 0
+    for _ in range(num_chunks):
+        msg = ch.try_receive()
+        if msg is None:
+            break
+        total += len(msg)
+    return total
+">>),
+
+    ChunkSizes = [256, 1024, 4096, 16384, 65536],
+    TotalBytes = 1048576,  % 1MB
+
+    io:format("~10s | ~14s | ~14s | ~8s~n",
+              ["Chunk", "Buffer (MB/s)", "Channel (MB/s)", "Ratio"]),
+    io:format("~s~n", [string:copies("-", 54)]),
+
+    lists:foreach(fun(ChunkSize) ->
+        NumChunks = TotalBytes div ChunkSize,
+        Chunk = binary:copy(<<0>>, ChunkSize),
+
+        %% Benchmark PyBuffer streaming (Erlang write -> Python read)
+        {ok, Buf} = py_buffer:new(TotalBytes),
+        BufStart = erlang:monotonic_time(microsecond),
+        lists:foreach(fun(_) ->
+            ok = py_buffer:write(Buf, Chunk)
+        end, lists:seq(1, NumChunks)),
+        ok = py_buffer:close(Buf),
+        {ok, _} = py:eval(Ctx, <<"stream_read_buffer(buf)">>, #{<<"buf">> => Buf}),
+        BufEnd = erlang:monotonic_time(microsecond),
+        BufTime = (BufEnd - BufStart) / 1000000,
+        BufMBPerSec = (TotalBytes / 1048576) / BufTime,
+
+        %% Benchmark Channel streaming (Erlang send -> Python receive)
+        {ok, Ch} = py_channel:new(),
+        ChanStart = erlang:monotonic_time(microsecond),
+        lists:foreach(fun(_) ->
+            ok = py_channel:send(Ch, Chunk)
+        end, lists:seq(1, NumChunks)),
+        {ok, _} = py:eval(Ctx, <<"stream_recv_channel(ch, num_chunks)">>,
+                          #{<<"ch">> => Ch, <<"num_chunks">> => NumChunks}),
+        ChanEnd = erlang:monotonic_time(microsecond),
+        ChanTime = (ChanEnd - ChanStart) / 1000000,
+        ChanMBPerSec = (TotalBytes / 1048576) / ChanTime,
+        py_channel:close(Ch),
+
+        Ratio = BufMBPerSec / ChanMBPerSec,
+        io:format("~10B | ~14.2f | ~14.2f | ~.2fx~n",
+                  [ChunkSize, BufMBPerSec, ChanMBPerSec, Ratio])
+    end, ChunkSizes),
+    ok.