Fix ReactorBuffer registration in subinterpreters and add benchmark

Author: benoitc

c_src/py_subinterp_pool.c

@@ -25,6 +25,7 @@
  */
 
 #include "py_subinterp_pool.h"
+#include "py_reactor_buffer.h"
 #include <string.h>
 
 #ifdef HAVE_SUBINTERPRETERS
@@ -163,6 +164,12 @@ int subinterp_pool_init(int size) {
             PyErr_Clear();
             /* Non-fatal - continue without erlang module */
         } else {
+            /* Register ReactorBuffer with erlang module in this subinterpreter */
+            if (ReactorBuffer_register_with_reactor() < 0) {
+                PyErr_Clear();
+                /* Non-fatal - ReactorBuffer just won't be available */
+            }
+
             /* Import erlang module into globals */
             PyObject *erlang_module = PyImport_ImportModule("erlang");
             if (erlang_module != NULL) {

examples/bench_reactor_buffer.erl

@@ -0,0 +1,287 @@
+#!/usr/bin/env escript
+%% -*- erlang -*-
+%%! -pa _build/default/lib/erlang_python/ebin
+
+%%% @doc Benchmark script for ReactorBuffer zero-copy performance.
+%%%
+%%% Run with:
+%%%   rebar3 compile && escript examples/bench_reactor_buffer.erl
+
+-mode(compile).
+
+main(_Args) ->
+    io:format("~n========================================~n"),
+    io:format("ReactorBuffer Zero-Copy Benchmark~n"),
+    io:format("========================================~n~n"),
+
+    %% Start the application
+    {ok, _} = application:ensure_all_started(erlang_python),
+    {ok, _} = py:start_contexts(),
+
+    %% 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_buffer_operations_bench(),
+    run_protocol_simulation_bench(),
+    run_echo_protocol_bench(),
+
+    io:format("~n========================================~n"),
+    io:format("Benchmark Complete~n"),
+    io:format("========================================~n"),
+
+    halt(0).
+
+run_buffer_operations_bench() ->
+    io:format("~n--- Buffer Operations Benchmark ---~n"),
+    io:format("Iterations: 10000~n~n"),
+
+    Code = <<"
+import time
+import erlang
+
+def run_buffer_ops_bench(iterations=10000):
+    results = {}
+    sizes = [64, 256, 1024, 4096, 16384, 65536]
+
+    for size in sizes:
+        test_data = b'X' * size
+        buf = erlang.ReactorBuffer._test_create(test_data)
+        regular_bytes = bytes(test_data)
+
+        # 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
+
+        start = time.perf_counter()
+        for _ in range(iterations):
+            ba = bytearray()
+            ba.extend(regular_bytes)
+        extend_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] = {
+            'extend_buf': extend_buf_time * 1000,
+            'extend_bytes': extend_bytes_time * 1000,
+            'startswith_buf': startswith_buf_time * 1000,
+            'startswith_bytes': startswith_bytes_time * 1000,
+        }
+
+    return results
+
+_buffer_ops_results = run_buffer_ops_bench()
+">>,
+
+    ok = py:exec(Code),
+    {ok, Results} = py:eval(<<"_buffer_ops_results">>),
+
+    io:format("~8s | ~12s | ~12s | ~12s | ~8s~n",
+              ["Size", "Operation", "Buffer (ms)", "Bytes (ms)", "Ratio"]),
+    io:format("~s~n", [string:copies("-", 60)]),
+
+    Sizes = [64, 256, 1024, 4096, 16384, 65536],
+    lists:foreach(fun(Size) ->
+        Data = maps:get(Size, Results),
+
+        ExtBuf = maps:get(<<"extend_buf">>, Data),
+        ExtBytes = maps:get(<<"extend_bytes">>, Data),
+        ExtRatio = ExtBytes / max(ExtBuf, 0.001),
+
+        SwBuf = maps:get(<<"startswith_buf">>, Data),
+        SwBytes = maps:get(<<"startswith_bytes">>, Data),
+        SwRatio = SwBytes / max(SwBuf, 0.001),
+
+        io:format("~8B | ~12s | ~12.3f | ~12.3f | ~.2f x~n",
+                  [Size, "extend", ExtBuf, ExtBytes, ExtRatio]),
+        io:format("~8s | ~12s | ~12.3f | ~12.3f | ~.2f x~n",
+                  ["", "startswith", SwBuf, SwBytes, SwRatio])
+    end, Sizes),
+    ok.
+
+run_protocol_simulation_bench() ->
+    io:format("~n--- Protocol Simulation Benchmark ---~n"),
+    io:format("Iterations: 5000~n~n"),
+
+    Code = <<"
+import time
+import erlang
+
+def run_protocol_sim_bench(iterations=5000):
+    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]
+
+        buf = erlang.ReactorBuffer._test_create(test_data)
+        regular_bytes = bytes(test_data)
+
+        def parse_request(data):
+            if data.startswith(b'GET'):
+                method = 'GET'
+            elif data.startswith(b'POST'):
+                method = 'POST'
+            else:
+                method = 'OTHER'
+            pos = data.find(b'\\r\\n\\r\\n')
+            write_buf = bytearray()
+            write_buf.extend(data)
+            return len(write_buf)
+
+        # Benchmark with ReactorBuffer
+        start = time.perf_counter()
+        for _ in range(iterations):
+            _ = parse_request(buf)
+        buf_time = time.perf_counter() - start
+
+        # Benchmark with regular bytes
+        start = time.perf_counter()
+        for _ in range(iterations):
+            _ = parse_request(regular_bytes)
+        bytes_time = time.perf_counter() - start
+
+        results[size] = {
+            'buffer_ops_per_sec': iterations / buf_time,
+            'bytes_ops_per_sec': iterations / bytes_time,
+        }
+
+    return results
+
+_protocol_sim_results = run_protocol_sim_bench()
+">>,
+
+    ok = py:exec(Code),
+    {ok, Results} = py:eval(<<"_protocol_sim_results">>),
+
+    io:format("~8s | ~14s | ~14s | ~8s~n",
+              ["Size", "Buffer (ops/s)", "Bytes (ops/s)", "Speedup"]),
+    io:format("~s~n", [string:copies("-", 52)]),
+
+    Sizes = [64, 256, 1024, 4096, 16384, 65536],
+    SpeedupsList = lists:map(fun(Size) ->
+        Data = maps:get(Size, Results),
+
+        BufOps = maps:get(<<"buffer_ops_per_sec">>, Data),
+        BytesOps = maps:get(<<"bytes_ops_per_sec">>, Data),
+        Speedup = BufOps / max(BytesOps, 1),
+
+        io:format("~8B | ~14w | ~14w | ~.2f~n",
+                  [Size, round(BufOps), round(BytesOps), Speedup]),
+        {Size, Speedup}
+    end, Sizes),
+
+    %% Calculate average speedup for >= 1KB
+    LargeSpeedups = [S || {Size, S} <- SpeedupsList, Size >= 1024],
+    case LargeSpeedups of
+        [] -> ok;
+        _ ->
+            AvgSpeedup = lists:sum(LargeSpeedups) / length(LargeSpeedups),
+            Improvement = (AvgSpeedup - 1.0) * 100,
+            io:format("~nAverage speedup for payloads >= 1KB: ~.2f x~n", [AvgSpeedup]),
+            io:format("Performance improvement: ~.1f%~n", [Improvement])
+    end,
+    ok.
+
+run_echo_protocol_bench() ->
+    io:format("~n--- Echo Protocol Benchmark ---~n"),
+    io:format("Iterations: 200~n~n"),
+
+    Code = <<"
+import time
+import socket
+import statistics
+import erlang.reactor as reactor
+
+def run_echo_bench(iterations=200):
+    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'
+
+    results = {}
+    sizes = [64, 256, 1024, 4096, 16384]
+
+    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
+
+_echo_bench_results = run_echo_bench()
+">>,
+
+    ok = py:exec(Code),
+    {ok, Results} = py:eval(<<"_echo_bench_results">>),
+
+    io:format("~8s | ~10s | ~10s | ~10s | ~10s~n",
+              ["Size", "Avg (ms)", "P50 (ms)", "P95 (ms)", "Ops/sec"]),
+    io:format("~s~n", [string:copies("-", 56)]),
+
+    Sizes = [64, 256, 1024, 4096, 16384],
+    lists:foreach(fun(Size) ->
+        Data = maps:get(Size, Results),
+
+        AvgMs = maps:get(<<"avg_time_ms">>, Data),
+        P50Ms = maps:get(<<"p50_ms">>, Data),
+        P95Ms = maps:get(<<"p95_ms">>, Data),
+        OpsPerSec = maps:get(<<"ops_per_sec">>, Data),
+
+        io:format("~8B | ~10.3f | ~10.3f | ~10.3f | ~10w~n",
+                  [Size, AvgMs, P50Ms, P95Ms, round(OpsPerSec)])
+    end, Sizes),
+    ok.