%% 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 async Python execution using event loops.

%%%

%%% This module provides an async request pool that delegates to the event loop

%%% pool for efficient coroutine execution. It replaces the pthread+usleep

%%% polling model with event-driven execution using enif_select and erlang.send().

%%%

%%% The pool maintains API compatibility with the previous pthread-based

%%% implementation while providing significant performance improvements.

%%%

%%% @private

-module(py_async_pool).

-behaviour(gen_server).

-export([

start_link/0,

start_link/1,

request/1,

get_stats/0

]).

-export([

init/1,

handle_call/3,

handle_cast/2,

handle_info/2,

terminate/2

]).

-record(state, {

pending :: non_neg_integer(),

supported :: boolean()

}).

%%% ============================================================================

%%% API

%%% ============================================================================

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

start_link() ->

start_link(1).

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

start_link(_NumWorkers) ->

%% NumWorkers is now ignored - we use the event loop pool instead

gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).

%% @doc Submit an async request to be executed by the event loop pool.

-spec request(term()) -> ok.

request(Request) ->

gen_server:cast(?MODULE, {request, Request}).

%% @doc Get pool statistics.

-spec get_stats() -> map().

get_stats() ->

gen_server:call(?MODULE, get_stats).

%%% ============================================================================

%%% gen_server callbacks

%%% ============================================================================

init([]) ->

process_flag(trap_exit, true),

%% Check if event loop pool is available

case py_event_loop:get_loop() of

{ok, _LoopRef} ->

{ok, #state{pending = 0, supported = true}};

{error, _} ->

{ok, #state{pending = 0, supported = false}}

end.

handle_call(get_stats, _From, State) ->

Stats = #{

pending_requests => State#state.pending,

supported => State#state.supported,

backend => event_loop

},

{reply, Stats, State};

handle_call(_Request, _From, State) ->

{reply, {error, unknown_request}, State}.

handle_cast({request, Request}, #state{supported = false} = State) ->

{Ref, Caller, _Type} = extract_ref_caller(Request),

Caller ! {py_error, Ref, async_not_supported},

{noreply, State};

handle_cast({request, Request}, State) ->

case transform_request(Request) of

{ok, LoopRequest} ->

case py_event_loop:get_loop() of

{ok, LoopRef} ->

case py_event_loop:run_async(LoopRef, LoopRequest) of

ok ->

{noreply, State#state{pending = State#state.pending + 1}};

{error, Reason} ->

{Ref, Caller, _} = extract_ref_caller(Request),

Caller ! {py_error, Ref, Reason},

{noreply, State}

end;

{error, Reason} ->

{Ref, Caller, _} = extract_ref_caller(Request),

Caller ! {py_error, Ref, Reason},

{noreply, State}

end;

{error, Reason} ->

{Ref, Caller, _} = extract_ref_caller(Request),

Caller ! {py_error, Ref, Reason},

{noreply, State}

end;

handle_cast(_Msg, State) ->

{noreply, State}.

handle_info({async_result, _Ref, _Result}, State) ->

%% Result was sent directly to caller via erlang.send()

%% We just track pending count

{noreply, State#state{pending = max(0, State#state.pending - 1)}};

handle_info(_Info, State) ->

{noreply, State}.

terminate(_Reason, _State) ->

ok.

%%% ============================================================================

%%% Internal functions

%%% ============================================================================

%% @doc Transform the legacy request format to the new event loop format.

transform_request({async_call, Ref, Caller, Module, Func, Args, Kwargs}) ->

{ok, #{

ref => Ref,

caller => Caller,

module => Module,

func => Func,

args => Args,

kwargs => Kwargs

}};

transform_request({async_gather, Ref, Caller, Calls}) ->

%% For gather, we need to wrap in a special gather coroutine

%% For now, return an error - gather needs special handling

{error, {gather_not_implemented, Ref, Caller, Calls}};

transform_request({async_stream, Ref, Caller, Module, Func, Args, Kwargs}) ->

%% For stream, we need async generator support

%% For now, return an error - stream needs special handling

{error, {stream_not_implemented, Ref, Caller, Module, Func, Args, Kwargs}};

transform_request(Other) ->

{error, {unknown_request_type, Other}}.

%% @doc Extract ref and caller from different request types.

extract_ref_caller({async_call, Ref, Caller, _, _, _, _}) -> {Ref, Caller, async_call};

extract_ref_caller({async_gather, Ref, Caller, _}) -> {Ref, Caller, async_gather};

extract_ref_caller({async_stream, Ref, Caller, _, _, _, _}) -> {Ref, Caller, async_stream}.