%% 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 Erlang-native asyncio event loop manager. %% %% This module provides the high-level interface for using the Erlang-backed %% asyncio event loop. It manages the lifecycle of event loops and routers, %% and registers callback functions for Python to call. %% %% @private -module(py_event_loop). -behaviour(gen_server). %% API -export([ start_link/0, stop/0, get_loop/0, register_callbacks/0 ]). %% gen_server callbacks -export([ init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3 ]). -record(state, { loop_ref :: reference() | undefined, router_pid :: pid() | undefined }). %% ============================================================================ %% API %% ============================================================================ -spec start_link() -> {ok, pid()} | {error, term()}. start_link() -> gen_server:start_link({local, ?MODULE}, ?MODULE, [], []). -spec stop() -> ok. stop() -> gen_server:stop(?MODULE). -spec get_loop() -> {ok, reference()} | {error, not_started}. get_loop() -> gen_server:call(?MODULE, get_loop). %% @doc Register event loop callbacks for Python access. -spec register_callbacks() -> ok. register_callbacks() -> %% Register all event loop functions as callbacks py_callback:register(py_event_loop_new, fun cb_event_loop_new/1), py_callback:register(py_event_loop_destroy, fun cb_event_loop_destroy/1), py_callback:register(py_event_loop_set_router, fun cb_event_loop_set_router/1), py_callback:register(py_event_loop_wakeup, fun cb_event_loop_wakeup/1), py_callback:register(py_event_loop_add_reader, fun cb_add_reader/1), py_callback:register(py_event_loop_remove_reader, fun cb_remove_reader/1), py_callback:register(py_event_loop_add_writer, fun cb_add_writer/1), py_callback:register(py_event_loop_remove_writer, fun cb_remove_writer/1), py_callback:register(py_event_loop_call_later, fun cb_call_later/1), py_callback:register(py_event_loop_cancel_timer, fun cb_cancel_timer/1), py_callback:register(py_event_loop_poll_events, fun cb_poll_events/1), py_callback:register(py_event_loop_get_pending, fun cb_get_pending/1), py_callback:register(py_event_loop_dispatch_callback, fun cb_dispatch_callback/1), py_callback:register(py_event_loop_dispatch_timer, fun cb_dispatch_timer/1), ok. %% ============================================================================ %% gen_server callbacks %% ============================================================================ init([]) -> %% Register callbacks on startup register_callbacks(), %% Create and initialize the event loop immediately case py_nif:event_loop_new() of {ok, LoopRef} -> {ok, RouterPid} = py_event_router:start_link(LoopRef), ok = py_nif:event_loop_set_router(LoopRef, RouterPid), %% Set shared router for per-loop created loops %% All loops created via _loop_new() in Python will use this router ok = py_nif:set_shared_router(RouterPid), %% Make the event loop available to Python ok = py_nif:set_python_event_loop(LoopRef), %% Set ErlangEventLoop as the default asyncio policy ok = set_default_policy(), {ok, #state{loop_ref = LoopRef, router_pid = RouterPid}}; {error, Reason} -> {stop, {event_loop_init_failed, Reason}} end. %% @doc Set ErlangEventLoop as the default asyncio event loop policy. set_default_policy() -> PrivDir = code:priv_dir(erlang_python), Code = iolist_to_binary([ "import sys\n", "priv_dir = '", PrivDir, "'\n", "if priv_dir not in sys.path:\n", " sys.path.insert(0, priv_dir)\n", "from erlang_loop import get_event_loop_policy\n", "import asyncio\n", "asyncio.set_event_loop_policy(get_event_loop_policy())\n" ]), case py:exec(Code) of ok -> ok; {error, Reason} -> error_logger:warning_msg("Failed to set ErlangEventLoop policy: ~p~n", [Reason]), ok %% Non-fatal end. handle_call(get_loop, _From, #state{loop_ref = undefined} = State) -> %% Create event loop and router on demand case py_nif:event_loop_new() of {ok, LoopRef} -> {ok, RouterPid} = py_event_router:start_link(LoopRef), ok = py_nif:event_loop_set_router(LoopRef, RouterPid), %% Make the event loop available to Python ok = py_nif:set_python_event_loop(LoopRef), NewState = State#state{loop_ref = LoopRef, router_pid = RouterPid}, {reply, {ok, LoopRef}, NewState}; {error, _} = Error -> {reply, Error, State} end; handle_call(get_loop, _From, #state{loop_ref = LoopRef} = State) -> {reply, {ok, LoopRef}, State}; handle_call(_Request, _From, State) -> {reply, {error, unknown_request}, State}. handle_cast(_Msg, State) -> {noreply, State}. handle_info(_Info, State) -> {noreply, State}. terminate(_Reason, #state{loop_ref = LoopRef, router_pid = RouterPid}) -> %% Reset asyncio policy back to default before destroying the loop reset_default_policy(), %% Clean up router case RouterPid of undefined -> ok; Pid -> py_event_router:stop(Pid) end, %% Clean up event loop case LoopRef of undefined -> ok; Ref -> py_nif:event_loop_destroy(Ref) end, ok. %% @doc Reset asyncio back to the default event loop policy. reset_default_policy() -> Code = <<" import asyncio asyncio.set_event_loop_policy(None) ">>, catch py:exec(Code), ok. code_change(_OldVsn, State, _Extra) -> {ok, State}. %% ============================================================================ %% Callback implementations for Python %% ============================================================================ cb_event_loop_new([]) -> py_nif:event_loop_new(). cb_event_loop_destroy([LoopRef]) -> py_nif:event_loop_destroy(LoopRef). cb_event_loop_set_router([LoopRef, RouterPid]) -> py_nif:event_loop_set_router(LoopRef, RouterPid). cb_event_loop_wakeup([LoopRef]) -> py_nif:event_loop_wakeup(LoopRef). cb_add_reader([LoopRef, Fd, CallbackId]) -> py_nif:add_reader(LoopRef, Fd, CallbackId). cb_remove_reader([LoopRef, FdRef]) -> py_nif:remove_reader(LoopRef, FdRef). cb_add_writer([LoopRef, Fd, CallbackId]) -> py_nif:add_writer(LoopRef, Fd, CallbackId). cb_remove_writer([LoopRef, FdRef]) -> py_nif:remove_writer(LoopRef, FdRef). cb_call_later([LoopRef, DelayMs, CallbackId]) -> py_nif:call_later(LoopRef, DelayMs, CallbackId). cb_cancel_timer([LoopRef, TimerRef]) -> py_nif:cancel_timer(LoopRef, TimerRef). cb_poll_events([LoopRef, TimeoutMs]) -> py_nif:poll_events(LoopRef, TimeoutMs). cb_get_pending([LoopRef]) -> py_nif:get_pending(LoopRef). cb_dispatch_callback([LoopRef, CallbackId, Type]) -> py_nif:dispatch_callback(LoopRef, CallbackId, Type). cb_dispatch_timer([LoopRef, CallbackId]) -> py_nif:dispatch_timer(LoopRef, CallbackId).