erlang_python

#!/usr/bin/env escript
%% -*- erlang -*-
%%! -pa _build/default/lib/erlang_python/ebin

%%% @doc Benchmark: Channel.receive() (sync) vs Channel.async_receive() (asyncio)
%%%
%%% Compares Python-side synchronous blocking receive against asyncio async_receive.
%%%
%%% Run with:
%%%   rebar3 compile && escript examples/bench_channel_receive.erl

-mode(compile).

main(_Args) ->
    io:format("~n========================================~n"),
    io:format("Channel Receive Benchmark~n"),
    io:format("sync receive() vs async_receive()~n"),
    io:format("========================================~n~n"),

    {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~n", [PyVer]),

    %% Run benchmarks
    run_sync_receive_bench(),
    run_async_receive_bench(),
    run_comparison(),

    io:format("~n========================================~n"),
    io:format("Benchmark Complete~n"),
    io:format("========================================~n"),

    halt(0).

run_sync_receive_bench() ->
    io:format("--- Sync Channel.receive() Benchmark ---~n"),
    io:format("(Python blocking receive via erlang.call)~n~n"),

    Sizes = [64, 1024, 16384],
    Iterations = 2000,

    io:format("~8s | ~12s | ~12s~n", ["Size", "Throughput", "Avg (us)"]),
    io:format("~s~n", [string:copies("-", 38)]),

    Ctx = py:context(1),

    %% Define sync receiver function
    ok = py:exec(Ctx, <<"
from erlang import Channel

def sync_receive_n(ch_ref, n):
    ch = Channel(ch_ref)
    for _ in range(n):
        ch.receive()
    return n
">>),

    lists:foreach(fun(Size) ->
        {ok, Ch} = py_channel:new(),
        Data = binary:copy(<<0>>, Size),

        %% Fill channel
        lists:foreach(fun(_) ->
            ok = py_channel:send(Ch, Data)
        end, lists:seq(1, Iterations)),

        %% Time Python sync receive
        Start = erlang:monotonic_time(microsecond),
        {ok, Iterations} = py:eval(Ctx, <<"sync_receive_n(ch, n)">>,
                                    #{<<"ch">> => Ch, <<"n">> => Iterations}),
        End = erlang:monotonic_time(microsecond),

        TotalTime = (End - Start) / 1000000,
        AvgUs = (TotalTime / Iterations) * 1000000,
        Throughput = round(Iterations / TotalTime),

        io:format("~8B | ~12w | ~12.2f~n", [Size, Throughput, AvgUs]),

        py_channel:close(Ch)
    end, Sizes),
    ok.

run_async_receive_bench() ->
    io:format("~n--- Async Channel.async_receive() Benchmark ---~n"),
    io:format("(Python asyncio via erlang.run)~n~n"),

    Sizes = [64, 1024, 16384],
    Iterations = 2000,

    io:format("~8s | ~12s | ~12s~n", ["Size", "Throughput", "Avg (us)"]),
    io:format("~s~n", [string:copies("-", 38)]),

    Ctx = py:context(1),

    %% Define async receiver function
    ok = py:exec(Ctx, <<"
import erlang
from erlang import Channel

async def async_receive_n(ch_ref, n):
    ch = Channel(ch_ref)
    for _ in range(n):
        await ch.async_receive()
    return n

def run_async_receive(ch_ref, n):
    return erlang.run(async_receive_n(ch_ref, n))
">>),

    lists:foreach(fun(Size) ->
        {ok, Ch} = py_channel:new(),
        Data = binary:copy(<<0>>, Size),

        %% Fill channel
        lists:foreach(fun(_) ->
            ok = py_channel:send(Ch, Data)
        end, lists:seq(1, Iterations)),

        %% Time Python async receive
        Start = erlang:monotonic_time(microsecond),
        {ok, Iterations} = py:eval(Ctx, <<"run_async_receive(ch, n)">>,
                                    #{<<"ch">> => Ch, <<"n">> => Iterations}),
        End = erlang:monotonic_time(microsecond),

        TotalTime = (End - Start) / 1000000,
        AvgUs = (TotalTime / Iterations) * 1000000,
        Throughput = round(Iterations / TotalTime),

        io:format("~8B | ~12w | ~12.2f~n", [Size, Throughput, AvgUs]),

        py_channel:close(Ch)
    end, Sizes),
    ok.

run_comparison() ->
    io:format("~n--- Direct Comparison ---~n"),
    io:format("(1KB messages, 2000 iterations)~n~n"),

    Size = 1024,
    Iterations = 2000,

    Ctx = py:context(1),

    %% Define both functions
    ok = py:exec(Ctx, <<"
import erlang
from erlang import Channel

def sync_receive_n(ch_ref, n):
    ch = Channel(ch_ref)
    for _ in range(n):
        ch.receive()
    return n

async def async_receive_n(ch_ref, n):
    ch = Channel(ch_ref)
    for _ in range(n):
        await ch.async_receive()
    return n

def run_async_receive(ch_ref, n):
    return erlang.run(async_receive_n(ch_ref, n))
">>),

    io:format("~15s | ~12s | ~12s~n", ["Method", "Time (ms)", "Throughput"]),
    io:format("~s~n", [string:copies("-", 45)]),

    Data = binary:copy(<<0>>, Size),

    %% Sync receive benchmark
    {ok, SyncCh} = py_channel:new(),
    lists:foreach(fun(_) -> ok = py_channel:send(SyncCh, Data) end, lists:seq(1, Iterations)),
    SyncStart = erlang:monotonic_time(microsecond),
    {ok, Iterations} = py:eval(Ctx, <<"sync_receive_n(ch, n)">>,
                                #{<<"ch">> => SyncCh, <<"n">> => Iterations}),
    SyncEnd = erlang:monotonic_time(microsecond),
    SyncTime = (SyncEnd - SyncStart) / 1000,
    SyncThroughput = round(Iterations / (SyncTime / 1000)),
    io:format("~15s | ~12.2f | ~12w~n", ["sync receive", SyncTime, SyncThroughput]),
    py_channel:close(SyncCh),

    %% Async receive benchmark
    {ok, AsyncCh} = py_channel:new(),
    lists:foreach(fun(_) -> ok = py_channel:send(AsyncCh, Data) end, lists:seq(1, Iterations)),
    AsyncStart = erlang:monotonic_time(microsecond),
    {ok, Iterations} = py:eval(Ctx, <<"run_async_receive(ch, n)">>,
                                #{<<"ch">> => AsyncCh, <<"n">> => Iterations}),
    AsyncEnd = erlang:monotonic_time(microsecond),
    AsyncTime = (AsyncEnd - AsyncStart) / 1000,
    AsyncThroughput = round(Iterations / (AsyncTime / 1000)),
    io:format("~15s | ~12.2f | ~12w~n", ["async_receive", AsyncTime, AsyncThroughput]),
    py_channel:close(AsyncCh),

    %% NIF baseline (no Python)
    {ok, NifCh} = py_channel:new(),
    lists:foreach(fun(_) -> ok = py_channel:send(NifCh, Data) end, lists:seq(1, Iterations)),
    NifStart = erlang:monotonic_time(microsecond),
    receive_all_nif(NifCh, Iterations),
    NifEnd = erlang:monotonic_time(microsecond),
    NifTime = (NifEnd - NifStart) / 1000,
    NifThroughput = round(Iterations / (NifTime / 1000)),
    io:format("~15s | ~12.2f | ~12w~n", ["NIF baseline", NifTime, NifThroughput]),
    py_channel:close(NifCh),

    %% Summary
    io:format("~n"),
    io:format("sync receive is ~.1fx faster than async_receive~n", [AsyncTime / SyncTime]),
    io:format("NIF baseline is ~.1fx faster than sync receive~n", [SyncTime / NifTime]),
    ok.

receive_all_nif(_Ch, 0) -> ok;
receive_all_nif(Ch, N) ->
    {ok, _} = py_nif:channel_try_receive(Ch),
    receive_all_nif(Ch, N - 1).