Add async_pending action and reactor NIF support

benoitc · benoitc · commit 87303b54e158 · 2026-03-08T10:58:11.000+01:00
- Add handoff/2,3 API functions for FD handoff
- Add setup_code option to run Python code in context
- Add async_pending action handling for task-based ASGI
- Add {write_ready, Fd} message handling for async completion
- Register reactor NIFs in nif_funcs table
- Fix erlang.reactor module import path in NIFs
- Add async_pending_test to py_reactor_SUITE
diff --git a/c_src/py_event_loop.c b/c_src/py_event_loop.c
@@ -3088,7 +3088,7 @@ ERL_NIF_TERM nif_reactor_on_read_ready(ErlNifEnv *env, int argc,
     gil_guard_t guard = gil_acquire();
 
     /* Import erlang_reactor module */
-    PyObject *reactor_module = PyImport_ImportModule("erlang_reactor");
+    PyObject *reactor_module = PyImport_ImportModule("erlang.reactor");
     if (reactor_module == NULL) {
         PyErr_Clear();
         gil_release(guard);
@@ -3153,7 +3153,7 @@ ERL_NIF_TERM nif_reactor_on_write_ready(ErlNifEnv *env, int argc,
     gil_guard_t guard = gil_acquire();
 
     /* Import erlang_reactor module */
-    PyObject *reactor_module = PyImport_ImportModule("erlang_reactor");
+    PyObject *reactor_module = PyImport_ImportModule("erlang.reactor");
     if (reactor_module == NULL) {
         PyErr_Clear();
         gil_release(guard);
@@ -3231,7 +3231,7 @@ ERL_NIF_TERM nif_reactor_init_connection(ErlNifEnv *env, int argc,
     }
 
     /* Import erlang_reactor module */
-    PyObject *reactor_module = PyImport_ImportModule("erlang_reactor");
+    PyObject *reactor_module = PyImport_ImportModule("erlang.reactor");
     if (reactor_module == NULL) {
         Py_DECREF(client_info);
         PyErr_Clear();
@@ -3286,7 +3286,7 @@ ERL_NIF_TERM nif_reactor_close_fd(ErlNifEnv *env, int argc,
     if (fd >= 0) {
         gil_guard_t guard = gil_acquire();
 
-        PyObject *reactor_module = PyImport_ImportModule("erlang_reactor");
+        PyObject *reactor_module = PyImport_ImportModule("erlang.reactor");
         if (reactor_module != NULL) {
             PyObject *result = PyObject_CallMethod(reactor_module,
                                                     "close_connection", "i", fd);
diff --git a/c_src/py_event_loop.h b/c_src/py_event_loop.h
@@ -799,4 +799,72 @@ int create_default_event_loop(ErlNifEnv *env);
  */
 int init_subinterpreter_event_loop(ErlNifEnv *env);
 
+/* ============================================================================
+ * Reactor NIF Functions (Erlang-as-Reactor architecture)
+ * ============================================================================ */
+
+/**
+ * @brief Register a file descriptor for reactor monitoring
+ *
+ * NIF: reactor_register_fd(ContextRef, Fd, OwnerPid) -> {ok, FdRef} | {error, Reason}
+ */
+ERL_NIF_TERM nif_reactor_register_fd(ErlNifEnv *env, int argc,
+                                      const ERL_NIF_TERM argv[]);
+
+/**
+ * @brief Re-register for read events after a one-shot event
+ *
+ * NIF: reactor_reselect_read(FdRef) -> ok | {error, Reason}
+ */
+ERL_NIF_TERM nif_reactor_reselect_read(ErlNifEnv *env, int argc,
+                                        const ERL_NIF_TERM argv[]);
+
+/**
+ * @brief Register for write events
+ *
+ * NIF: reactor_select_write(FdRef) -> ok | {error, Reason}
+ */
+ERL_NIF_TERM nif_reactor_select_write(ErlNifEnv *env, int argc,
+                                       const ERL_NIF_TERM argv[]);
+
+/**
+ * @brief Get the FD integer from an FD resource
+ *
+ * NIF: get_fd_from_resource(FdRef) -> Fd | {error, Reason}
+ */
+ERL_NIF_TERM nif_get_fd_from_resource(ErlNifEnv *env, int argc,
+                                       const ERL_NIF_TERM argv[]);
+
+/**
+ * @brief Call Python protocol on_read_ready
+ *
+ * NIF: reactor_on_read_ready(ContextRef, Fd) -> {ok, Action} | {error, Reason}
+ */
+ERL_NIF_TERM nif_reactor_on_read_ready(ErlNifEnv *env, int argc,
+                                        const ERL_NIF_TERM argv[]);
+
+/**
+ * @brief Call Python protocol on_write_ready
+ *
+ * NIF: reactor_on_write_ready(ContextRef, Fd) -> {ok, Action} | {error, Reason}
+ */
+ERL_NIF_TERM nif_reactor_on_write_ready(ErlNifEnv *env, int argc,
+                                         const ERL_NIF_TERM argv[]);
+
+/**
+ * @brief Initialize connection with Python protocol
+ *
+ * NIF: reactor_init_connection(ContextRef, Fd, ClientInfo) -> ok | {error, Reason}
+ */
+ERL_NIF_TERM nif_reactor_init_connection(ErlNifEnv *env, int argc,
+                                          const ERL_NIF_TERM argv[]);
+
+/**
+ * @brief Close FD and cleanup Python protocol
+ *
+ * NIF: reactor_close_fd(FdRef) -> ok | {error, Reason}
+ */
+ERL_NIF_TERM nif_reactor_close_fd(ErlNifEnv *env, int argc,
+                                   const ERL_NIF_TERM argv[]);
+
 #endif /* PY_EVENT_LOOP_H */
diff --git a/c_src/py_nif.c b/c_src/py_nif.c
@@ -39,6 +39,7 @@
 #include "py_nif.h"
 #include "py_asgi.h"
 #include "py_wsgi.h"
+#include "py_event_loop.h"
 
 /* ============================================================================
  * Global state definitions
@@ -3865,7 +3866,17 @@ static ErlNifFunc nif_funcs[] = {
     {"ref_interp_id", 1, nif_ref_interp_id, 0},
     {"ref_to_term", 1, nif_ref_to_term, 0},
     {"ref_getattr", 2, nif_ref_getattr, ERL_NIF_DIRTY_JOB_CPU_BOUND},
-    {"ref_call_method", 3, nif_ref_call_method, ERL_NIF_DIRTY_JOB_CPU_BOUND}
+    {"ref_call_method", 3, nif_ref_call_method, ERL_NIF_DIRTY_JOB_CPU_BOUND},
+
+    /* Reactor NIFs - Erlang-as-Reactor architecture */
+    {"reactor_register_fd", 3, nif_reactor_register_fd, 0},
+    {"reactor_reselect_read", 1, nif_reactor_reselect_read, 0},
+    {"reactor_select_write", 1, nif_reactor_select_write, 0},
+    {"get_fd_from_resource", 1, nif_get_fd_from_resource, 0},
+    {"reactor_on_read_ready", 2, nif_reactor_on_read_ready, ERL_NIF_DIRTY_JOB_CPU_BOUND},
+    {"reactor_on_write_ready", 2, nif_reactor_on_write_ready, ERL_NIF_DIRTY_JOB_CPU_BOUND},
+    {"reactor_init_connection", 3, nif_reactor_init_connection, ERL_NIF_DIRTY_JOB_CPU_BOUND},
+    {"reactor_close_fd", 1, nif_reactor_close_fd, 0}
 };
 
 ERL_NIF_INIT(py_nif, nif_funcs, load, NULL, upgrade, unload)
diff --git a/src/py_reactor_context.erl b/src/py_reactor_context.erl
@@ -40,7 +40,9 @@
     start_link/2,
     start_link/3,
     stop/1,
-    stats/1
+    stats/1,
+    handoff/2,
+    handoff/3
 ]).
 
 %% Internal exports
@@ -89,6 +91,8 @@ start_link(Id, Mode) ->
 %% - max_connections: Maximum connections per context (default: 100)
 %% - app_module: Python module containing ASGI/WSGI app
 %% - app_callable: Python callable name (e.g., "app", "application")
+%% - setup_code: Binary Python code to execute after context creation
+%%               (useful for setting up protocol factory)
 %%
 %% @param Id Unique identifier for this context
 %% @param Mode Context mode (auto, subinterp, worker)
@@ -141,6 +145,33 @@ stats(Ctx) when is_pid(Ctx) ->
         {error, timeout}
     end.
 
+%% @doc Hand off a file descriptor to this reactor context.
+%%
+%% The context takes ownership of the FD and will handle I/O events.
+%% This is the main entry point for handing off accepted connections.
+%%
+%% @param Fd The raw file descriptor (from inet:getfd/1)
+%% @param ClientInfo Map with connection metadata (addr, port, type, etc.)
+-spec handoff(integer(), map()) -> ok | {error, term()}.
+handoff(Fd, ClientInfo) when is_integer(Fd), is_map(ClientInfo) ->
+    %% Get a reactor context from the pool or use default
+    case whereis(py_reactor_context_default) of
+        undefined ->
+            {error, no_reactor_context};
+        Ctx ->
+            handoff(Ctx, Fd, ClientInfo)
+    end.
+
+%% @doc Hand off a file descriptor to a specific reactor context.
+%%
+%% @param Ctx The reactor context pid
+%% @param Fd The raw file descriptor (from inet:getfd/1)
+%% @param ClientInfo Map with connection metadata
+-spec handoff(pid(), integer(), map()) -> ok | {error, term()}.
+handoff(Ctx, Fd, ClientInfo) when is_pid(Ctx), is_integer(Fd), is_map(ClientInfo) ->
+    Ctx ! {fd_handoff, Fd, ClientInfo},
+    ok.
+
 %% ============================================================================
 %% Process loop
 %% ============================================================================
@@ -169,6 +200,19 @@ init(Parent, Id, Mode, Opts) ->
             AppModule = maps:get(app_module, Opts, undefined),
             AppCallable = maps:get(app_callable, Opts, undefined),
 
+            %% Execute setup code if specified (e.g., set protocol factory)
+            SetupCode = maps:get(setup_code, Opts, undefined),
+            case SetupCode of
+                undefined -> ok;
+                _ when is_binary(SetupCode) ->
+                    case py_nif:context_exec(Ref, SetupCode) of
+                        ok -> ok;
+                        {error, Reason} ->
+                            error_logger:error_msg(
+                                "py_reactor_context setup_code failed: ~p~n", [Reason])
+                    end
+            end,
+
             %% Initialize app in Python context if specified
             case AppModule of
                 undefined -> ok;
@@ -214,6 +258,15 @@ loop(State) ->
         {select, FdRes, _Ref, ready_output} ->
             handle_write_ready(FdRes, State);
 
+        %% Async completion signal from Python
+        %% Sent when an async task (e.g., ASGI app) completes and response is ready
+        %% Accept both atom and binary forms since Python sends binaries
+        {write_ready, Fd} ->
+            handle_async_write_ready(Fd, State);
+
+        {<<"write_ready">>, Fd} ->
+            handle_async_write_ready(Fd, State);
+
         %% Control messages
         {stop, From, MRef} ->
             cleanup(State),
@@ -265,8 +318,11 @@ handle_fd_handoff(Fd, ClientInfo, State) ->
             %% Register FD for monitoring
             case py_nif:reactor_register_fd(Ref, Fd, self()) of
                 {ok, FdRef} ->
+                    %% Inject reactor_pid into client_info for async signaling
+                    ClientInfoWithPid = ClientInfo#{reactor_pid => self()},
+
                     %% Initialize Python protocol handler
-                    case py_nif:reactor_init_connection(Ref, Fd, ClientInfo) of
+                    case py_nif:reactor_init_connection(Ref, Fd, ClientInfoWithPid) of
                         ok ->
                             %% Store connection info
                             ConnInfo = #{
@@ -319,6 +375,16 @@ handle_read_ready(FdRes, State) ->
                     },
                     loop(NewState);
 
+                {ok, <<"async_pending">>} ->
+                    %% Async task submitted (e.g., ASGI app running as task)
+                    %% Don't reselect - wait for {write_ready, Fd} signal from Python
+                    %% Increment request count since task was accepted
+                    error_logger:info_msg("Received async_pending for fd=~p~n", [Fd]),
+                    NewState = State#state{
+                        total_requests = State#state.total_requests + 1
+                    },
+                    loop(NewState);
+
                 {ok, <<"close">>} ->
                     %% Close connection
                     close_connection(Fd, FdRes, State);
@@ -370,6 +436,31 @@ handle_write_ready(FdRes, State) ->
             loop(State)
     end.
 
+%% ============================================================================
+%% Async Write Ready Handler
+%% ============================================================================
+
+%% @private
+%% Handle async completion signal from Python.
+%% This is sent when an async task (like an ASGI app) has completed
+%% and the response buffer is ready to be written.
+handle_async_write_ready(Fd, State) ->
+    #state{connections = Conns} = State,
+
+    error_logger:info_msg("handle_async_write_ready called for fd=~p~n", [Fd]),
+
+    case maps:get(Fd, Conns, undefined) of
+        #{fd_ref := FdRef} ->
+            %% Response buffer is ready, trigger write selection
+            error_logger:info_msg("Triggering write selection for fd=~p~n", [Fd]),
+            py_nif:reactor_select_write(FdRef),
+            loop(State);
+        undefined ->
+            %% Connection not found (may have been closed), ignore
+            error_logger:warning_msg("Connection not found for fd=~p~n", [Fd]),
+            loop(State)
+    end.
+
 %% ============================================================================
 %% Connection Management
 %% ============================================================================
diff --git a/test/py_reactor_SUITE.erl b/test/py_reactor_SUITE.erl
