erlang_python

%% Copyright 2026 Benoit Chesneau
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%%     http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.

%%% @doc Pool manager for event loop-based async Python execution.
%%%
%%% This module provides a pool of event loops for executing async Python
%%% coroutines. It replaces the pthread+usleep polling model with efficient
%%% event-driven execution using enif_select and erlang.send().
%%%
%%% The pool uses round-robin scheduling to distribute work across event loops.
%%%
%%% @private
-module(py_event_loop_pool).
-behaviour(gen_server).

-export([
    start_link/0,
    start_link/1,
    run_async/1,
    get_stats/0
]).

-export([
    init/1,
    handle_call/3,
    handle_cast/2,
    handle_info/2,
    terminate/2
]).

-record(state, {
    loops :: tuple(),  %% tuple of {LoopRef, WorkerPid} for O(1) access
    num_loops :: non_neg_integer(),
    next_idx :: non_neg_integer(),
    supported :: boolean()
}).

-define(DEFAULT_NUM_LOOPS, 1).

%%% ============================================================================
%%% API
%%% ============================================================================

-spec start_link() -> {ok, pid()} | {error, term()}.
start_link() ->
    start_link(?DEFAULT_NUM_LOOPS).

-spec start_link(pos_integer()) -> {ok, pid()} | {error, term()}.
start_link(NumLoops) ->
    gen_server:start_link({local, ?MODULE}, ?MODULE, [NumLoops], []).

%% @doc Submit an async request to be executed on the event loop pool.
%% The request should be a map with keys:
%%   ref => reference() - A reference to identify the result
%%   caller => pid() - The pid to send the result to
%%   module => atom() | binary() - Python module name
%%   func => atom() | binary() - Python function name
%%   args => list() - Arguments to pass to the function
%%   kwargs => map() - Keyword arguments (optional)
-spec run_async(map()) -> ok | {error, term()}.
run_async(Request) ->
    gen_server:call(?MODULE, {run_async, Request}).

%% @doc Get pool statistics.
-spec get_stats() -> map().
get_stats() ->
    gen_server:call(?MODULE, get_stats).

%%% ============================================================================
%%% gen_server callbacks
%%% ============================================================================

init([NumLoops]) ->
    process_flag(trap_exit, true),

    %% Create multiple independent event loops for true parallelism
    case create_loops(NumLoops, []) of
        {ok, LoopList} ->
            %% Convert to tuple for O(1) element access
            Loops = list_to_tuple(LoopList),
            {ok, #state{
                loops = Loops,
                num_loops = NumLoops,
                next_idx = 0,
                supported = true
            }};
        {error, Reason} ->
            error_logger:warning_msg("py_event_loop_pool: failed to create loops: ~p~n", [Reason]),
            {ok, #state{
                loops = {},
                num_loops = 0,
                next_idx = 0,
                supported = false
            }}
    end.

%% @private Create NumLoops independent event loops with workers
create_loops(0, Acc) ->
    {ok, lists:reverse(Acc)};
create_loops(N, Acc) ->
    case py_nif:event_loop_new() of
        {ok, LoopRef} ->
            WorkerId = iolist_to_binary([<<"pool_">>, integer_to_binary(N)]),
            case py_event_worker:start_link(WorkerId, LoopRef) of
                {ok, WorkerPid} ->
                    ok = py_nif:event_loop_set_worker(LoopRef, WorkerPid),
                    ok = py_nif:event_loop_set_id(LoopRef, WorkerId),
                    create_loops(N - 1, [{LoopRef, WorkerPid} | Acc]);
                {error, Reason} ->
                    {error, {worker_start_failed, Reason}}
            end;
        {error, Reason} ->
            {error, {loop_create_failed, Reason}}
    end.

handle_call(get_stats, _From, State) ->
    Stats = #{
        num_loops => State#state.num_loops,
        next_idx => State#state.next_idx,
        supported => State#state.supported
    },
    {reply, Stats, State};

handle_call({run_async, _Request}, _From, #state{supported = false} = State) ->
    {reply, {error, event_loop_not_available}, State};

handle_call({run_async, Request}, _From, State) ->
    %% Get the next loop in round-robin fashion with O(1) tuple access
    Idx = State#state.next_idx rem State#state.num_loops + 1,
    {LoopRef, _WorkerPid} = element(Idx, State#state.loops),

    %% Submit to the event loop
    Result = py_event_loop:run_async(LoopRef, Request),

    NextState = State#state{next_idx = Idx},
    {reply, Result, NextState};

handle_call(_Request, _From, State) ->
    {reply, {error, unknown_request}, State}.

handle_cast(_Msg, State) ->
    {noreply, State}.

handle_info({'EXIT', _Pid, _Reason}, State) ->
    %% A worker died, mark pool as unsupported
    %% Future: could try to restart just the failed loop/worker pair
    {noreply, State#state{supported = false, loops = {}, num_loops = 0}};

handle_info(_Info, State) ->
    {noreply, State}.

terminate(_Reason, #state{loops = Loops}) ->
    %% Clean up all loops and workers
    lists:foreach(fun({LoopRef, WorkerPid}) ->
        catch py_event_worker:stop(WorkerPid),
        catch py_nif:event_loop_destroy(LoopRef)
    end, tuple_to_list(Loops)),
    ok.