Add Channel API benchmark comparing with Reactor

Author: benoitc

examples/bench_channel.erl

@@ -0,0 +1,269 @@
+#!/usr/bin/env escript
+%% -*- erlang -*-
+%%! -pa _build/default/lib/erlang_python/ebin
+
+%%% @doc Benchmark script for Channel API vs Reactor comparison.
+%%%
+%%% Run with:
+%%%   rebar3 compile && escript examples/bench_channel.erl
+
+-mode(compile).
+
+main(_Args) ->
+    io:format("~n========================================~n"),
+    io:format("Channel API 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_channel_throughput_bench(),
+    run_channel_vs_reactor_bench(),
+    run_channel_latency_bench(),
+
+    io:format("~n========================================~n"),
+    io:format("Benchmark Complete~n"),
+    io:format("========================================~n"),
+
+    halt(0).
+
+run_channel_throughput_bench() ->
+    io:format("~n--- Channel Throughput Benchmark ---~n"),
+    io:format("Messages per batch: 1000~n~n"),
+
+    Sizes = [64, 256, 1024, 4096, 16384],
+
+    io:format("~8s | ~12s | ~12s | ~12s~n",
+              ["Size", "Send (msg/s)", "Recv (msg/s)", "Round (msg/s)"]),
+    io:format("~s~n", [string:copies("-", 52)]),
+
+    lists:foreach(fun(Size) ->
+        {ok, Ch} = py_channel:new(),
+        Data = binary:copy(<<0>>, Size),
+        Iterations = 1000,
+
+        %% Benchmark send
+        SendStart = erlang:monotonic_time(microsecond),
+        lists:foreach(fun(_) ->
+            ok = py_channel:send(Ch, Data)
+        end, lists:seq(1, Iterations)),
+        SendEnd = erlang:monotonic_time(microsecond),
+        SendTime = (SendEnd - SendStart) / 1000000,
+        SendRate = Iterations / SendTime,
+
+        %% Benchmark receive (drain the queue)
+        RecvStart = erlang:monotonic_time(microsecond),
+        lists:foreach(fun(_) ->
+            {ok, _} = py_nif:channel_try_receive(Ch)
+        end, lists:seq(1, Iterations)),
+        RecvEnd = erlang:monotonic_time(microsecond),
+        RecvTime = (RecvEnd - RecvStart) / 1000000,
+        RecvRate = Iterations / RecvTime,
+
+        %% Benchmark round-trip (send + receive)
+        RoundStart = erlang:monotonic_time(microsecond),
+        lists:foreach(fun(_) ->
+            ok = py_channel:send(Ch, Data),
+            {ok, _} = py_nif:channel_try_receive(Ch)
+        end, lists:seq(1, Iterations)),
+        RoundEnd = erlang:monotonic_time(microsecond),
+        RoundTime = (RoundEnd - RoundStart) / 1000000,
+        RoundRate = Iterations / RoundTime,
+
+        io:format("~8B | ~12w | ~12w | ~12w~n",
+                  [Size, round(SendRate), round(RecvRate), round(RoundRate)]),
+
+        py_channel:close(Ch)
+    end, Sizes),
+    ok.
+
+run_channel_vs_reactor_bench() ->
+    io:format("~n--- Channel vs Reactor Comparison ---~n"),
+    io:format("Iterations: 5000~n~n"),
+
+    Code = <<"
+import time
+import socket
+import erlang
+import erlang.reactor as reactor
+
+def bench_channel_python(iterations=5000):
+    '''Benchmark Channel from Python side'''
+    results = {}
+    sizes = [64, 256, 1024, 4096, 16384]
+
+    for size in sizes:
+        test_data = b'X' * size
+
+        # We can't easily create channels from Python yet,
+        # so we benchmark try_receive on pre-filled channels
+        # This measures Python-side overhead
+
+        # Measure try_receive with None ref (measures call overhead)
+        start = time.perf_counter()
+        for _ in range(iterations):
+            try:
+                # Just measure the erlang.call overhead
+                pass
+            except:
+                pass
+        call_time = time.perf_counter() - start
+
+        results[size] = {
+            'call_overhead_ms': call_time * 1000,
+            'ops_per_sec': iterations / max(call_time, 0.0001),
+        }
+
+    return results
+
+def bench_reactor_protocol(iterations=200):
+    '''Benchmark Reactor Protocol pattern'''
+    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()
+
+        import statistics
+        avg_time = statistics.mean(times)
+        results[size] = {
+            'avg_time_ms': avg_time * 1000,
+            'ops_per_sec': 1.0 / avg_time,
+        }
+
+    return results
+
+_reactor_results = bench_reactor_protocol()
+">>,
+
+    ok = py:exec(Code),
+    {ok, ReactorResults} = py:eval(<<"_reactor_results">>),
+
+    io:format("Reactor Protocol (echo pattern):~n"),
+    io:format("~8s | ~12s | ~12s~n",
+              ["Size", "Avg (ms)", "Ops/sec"]),
+    io:format("~s~n", [string:copies("-", 36)]),
+
+    Sizes = [64, 256, 1024, 4096, 16384],
+    lists:foreach(fun(Size) ->
+        Data = maps:get(Size, ReactorResults),
+        AvgMs = maps:get(<<"avg_time_ms">>, Data),
+        OpsPerSec = maps:get(<<"ops_per_sec">>, Data),
+        io:format("~8B | ~12.3f | ~12w~n",
+                  [Size, AvgMs, round(OpsPerSec)])
+    end, Sizes),
+
+    %% Now compare with Channel round-trip
+    io:format("~nChannel Round-trip (Erlang send + NIF receive):~n"),
+    io:format("~8s | ~12s | ~12s | ~8s~n",
+              ["Size", "Avg (ms)", "Ops/sec", "vs React"]),
+    io:format("~s~n", [string:copies("-", 48)]),
+
+    lists:foreach(fun(Size) ->
+        {ok, Ch} = py_channel:new(),
+        Data = binary:copy(<<0>>, Size),
+        Iterations = 1000,
+
+        Start = erlang:monotonic_time(microsecond),
+        lists:foreach(fun(_) ->
+            ok = py_channel:send(Ch, Data),
+            {ok, _} = py_nif:channel_try_receive(Ch)
+        end, lists:seq(1, Iterations)),
+        End = erlang:monotonic_time(microsecond),
+
+        TotalTime = (End - Start) / 1000000,
+        AvgMs = (TotalTime / Iterations) * 1000,
+        OpsPerSec = Iterations / TotalTime,
+
+        ReactorData = maps:get(Size, ReactorResults),
+        ReactorOps = maps:get(<<"ops_per_sec">>, ReactorData),
+        Ratio = OpsPerSec / ReactorOps,
+
+        io:format("~8B | ~12.3f | ~12w | ~.1fx~n",
+                  [Size, AvgMs, round(OpsPerSec), Ratio]),
+
+        py_channel:close(Ch)
+    end, Sizes),
+    ok.
+
+run_channel_latency_bench() ->
+    io:format("~n--- Channel Latency Distribution ---~n"),
+    io:format("Iterations: 10000~n~n"),
+
+    Sizes = [64, 1024, 16384],
+
+    io:format("~8s | ~10s | ~10s | ~10s | ~10s~n",
+              ["Size", "Min (us)", "Avg (us)", "P99 (us)", "Max (us)"]),
+    io:format("~s~n", [string:copies("-", 56)]),
+
+    lists:foreach(fun(Size) ->
+        {ok, Ch} = py_channel:new(),
+        Data = binary:copy(<<0>>, Size),
+        Iterations = 10000,
+
+        %% Collect latencies
+        Latencies = lists:map(fun(_) ->
+            Start = erlang:monotonic_time(microsecond),
+            ok = py_channel:send(Ch, Data),
+            {ok, _} = py_nif:channel_try_receive(Ch),
+            End = erlang:monotonic_time(microsecond),
+            End - Start
+        end, lists:seq(1, Iterations)),
+
+        Sorted = lists:sort(Latencies),
+        Min = hd(Sorted),
+        Max = lists:last(Sorted),
+        Avg = lists:sum(Latencies) / Iterations,
+        P99Idx = round(Iterations * 0.99),
+        P99 = lists:nth(P99Idx, Sorted),
+
+        io:format("~8B | ~10.1f | ~10.1f | ~10.1f | ~10.1f~n",
+                  [Size, float(Min), Avg, float(P99), float(Max)]),
+
+        py_channel:close(Ch)
+    end, Sizes),
+    ok.