Add erlang.async_call() for asyncio-compatible callbacks

Implements a new async API that integrates properly with asyncio without
raising exceptions for control flow. This is the recommended approach for
ASGI/asyncio applications (FastAPI, Starlette, etc).

Usage:
    result = await erlang.async_call('func', arg1, arg2)

The implementation:
- Creates an asyncio Future for each call
- Sends request to Erlang via thread coordinator
- Releases the dirty NIF thread while waiting
- Uses a non-blocking pipe monitored by asyncio
- Resolves the Future when Erlang responds

This avoids the SuspensionRequired exception that conflicts with asyncio's
Task error handling when middleware catches and re-raises exceptions.

Author: benoitc

c_src/py_callback.c

@@ -1765,7 +1765,10 @@ static ErlNifFunc nif_funcs[] = {
     /* Thread worker support (ThreadPoolExecutor) */
     {"thread_worker_set_coordinator", 1, nif_thread_worker_set_coordinator, 0},
     {"thread_worker_write", 2, nif_thread_worker_write, 0},
-    {"thread_worker_signal_ready", 1, nif_thread_worker_signal_ready, 0}
+    {"thread_worker_signal_ready", 1, nif_thread_worker_signal_ready, 0},
+
+    /* Async callback support (for erlang.async_call) */
+    {"async_callback_response", 3, nif_async_callback_response, 0}
 };
 
 ERL_NIF_INIT(py_nif, nif_funcs, load, NULL, upgrade, unload)

c_src/py_nif.c

@@ -89,6 +89,8 @@
 #include <time.h>
 #include <math.h>
 #include <unistd.h>
+#include <fcntl.h>
+#include <errno.h>
 #include <pthread.h>
 
 #if defined(__linux__) || defined(__FreeBSD__) || defined(__NetBSD__) || defined(__OpenBSD__)

c_src/py_nif.h

@@ -104,10 +104,10 @@ static bool g_thread_worker_key_created = false;
 static __thread thread_worker_t *tl_thread_worker = NULL;
 
 /** @brief PID of the Erlang thread coordinator process */
-static ErlNifPid g_thread_coordinator_pid;
+ErlNifPid g_thread_coordinator_pid;
 
 /** @brief Flag: coordinator PID has been set */
-static bool g_has_thread_coordinator = false;
+bool g_has_thread_coordinator = false;
 
 /* ============================================================================
  * Thread Worker Destructor (pthread_key cleanup)
@@ -550,3 +550,54 @@ static ERL_NIF_TERM nif_thread_worker_signal_ready(ErlNifEnv *env, int argc,
 
     return ATOM_OK;
 }
+
+/**
+ * @brief NIF to write an async callback response
+ *
+ * Args: WriteFd, CallbackId, ResponseBinary
+ * Called by Erlang to send the result of an async callback back to Python.
+ * The response is written to the async callback pipe in the format:
+ *   callback_id (8 bytes) + response_len (4 bytes) + response_data
+ */
+static ERL_NIF_TERM nif_async_callback_response(ErlNifEnv *env, int argc,
+                                                 const ERL_NIF_TERM argv[]) {
+    (void)argc;
+    int fd;
+    ErlNifUInt64 callback_id;
+    ErlNifBinary response;
+
+    if (!enif_get_int(env, argv[0], &fd)) {
+        return make_error(env, "invalid_fd");
+    }
+
+    if (!enif_get_uint64(env, argv[1], &callback_id)) {
+        return make_error(env, "invalid_callback_id");
+    }
+
+    if (!enif_inspect_binary(env, argv[2], &response)) {
+        return make_error(env, "invalid_response");
+    }
+
+    /* Write callback_id (8 bytes) */
+    ssize_t n = write(fd, &callback_id, sizeof(callback_id));
+    if (n != sizeof(callback_id)) {
+        return make_error(env, "write_callback_id_failed");
+    }
+
+    /* Write response_len (4 bytes) */
+    uint32_t len = (uint32_t)response.size;
+    n = write(fd, &len, sizeof(len));
+    if (n != sizeof(len)) {
+        return make_error(env, "write_length_failed");
+    }
+
+    /* Write response_data */
+    if (len > 0) {
+        n = write(fd, response.data, response.size);
+        if (n != (ssize_t)response.size) {
+            return make_error(env, "write_data_failed");
+        }
+    }
+
+    return ATOM_OK;
+}

c_src/py_thread_worker.c

@@ -63,7 +63,9 @@
     %% Thread worker support (ThreadPoolExecutor)
     thread_worker_set_coordinator/1,
     thread_worker_write/2,
-    thread_worker_signal_ready/1
+    thread_worker_signal_ready/1,
+    %% Async callback support (for erlang.async_call)
+    async_callback_response/3
 ]).
 
 -on_load(load_nif/0).
@@ -382,3 +384,20 @@ thread_worker_write(_Fd, _Response) ->
 -spec thread_worker_signal_ready(integer()) -> ok | {error, term()}.
 thread_worker_signal_ready(_Fd) ->
     ?NIF_STUB.
+
+%%% ============================================================================
+%%% Async Callback Support (for erlang.async_call)
+%%% ============================================================================
+
+%% @doc Write an async callback response to the async callback pipe.
+%% Fd is the write end of the async callback pipe.
+%% CallbackId is the unique identifier for this callback.
+%% Response is the result binary (status byte + encoded result).
+%%
+%% This is called when an async_callback message is processed.
+%% The response is written in the format:
+%%   callback_id (8 bytes) + response_len (4 bytes) + response_data
+-spec async_callback_response(integer(), non_neg_integer(), binary()) ->
+    ok | {error, term()}.
+async_callback_response(_Fd, _CallbackId, _Response) ->
+    ?NIF_STUB.

src/py_nif.erl

@@ -105,6 +105,17 @@ handle_info({thread_callback, WorkerId, CallbackId, FuncName, Args},
     end,
     {noreply, State};
 
+%% Handle async callback request from Python async_call()
+%% Unlike thread_callback, this uses a global pipe for all async callbacks.
+%% Each response includes the callback_id so Python can match it to the right Future.
+handle_info({async_callback, CallbackId, FuncName, Args, WriteFd}, State) ->
+    %% Spawn a process to handle this callback asynchronously
+    %% This allows multiple async callbacks to be processed concurrently
+    spawn_link(fun() ->
+        handle_async_callback(WriteFd, CallbackId, FuncName, Args)
+    end),
+    {noreply, State};
+
 %% Handle handler process exit
 handle_info({'EXIT', Pid, _Reason}, #state{handlers = Handlers} = State) ->
     %% Remove handler from map
@@ -172,6 +183,37 @@ handle_thread_callback(WriteFd, FuncName, Args) ->
     %% Write response to pipe
     py_nif:thread_worker_write(WriteFd, Response).
 
+%% Execute an async callback and write response to the async callback pipe.
+%% Unlike handle_thread_callback, this includes the callback_id in the response
+%% so Python can match it to the correct Future.
+handle_async_callback(WriteFd, CallbackId, FuncName, Args) ->
+    %% Convert Args from tuple to list if needed
+    ArgsList = case Args of
+        T when is_tuple(T) -> tuple_to_list(T);
+        L when is_list(L) -> L;
+        _ -> [Args]
+    end,
+
+    %% Execute the registered function
+    Response = case py_callback:execute(FuncName, ArgsList) of
+        {ok, Result} ->
+            %% Encode result as Python-parseable string
+            %% Format: status_byte (0=ok) + python_repr
+            ResultStr = term_to_python_repr(Result),
+            <<0, ResultStr/binary>>;
+        {error, {not_found, Name}} ->
+            ErrMsg = iolist_to_binary(
+                io_lib:format("Function '~s' not registered", [Name])),
+            <<1, ErrMsg/binary>>;
+        {error, {Class, Reason, _Stack}} ->
+            ErrMsg = iolist_to_binary(
+                io_lib:format("~p: ~p", [Class, Reason])),
+            <<1, ErrMsg/binary>>
+    end,
+
+    %% Write response to async callback pipe (includes callback_id)
+    py_nif:async_callback_response(WriteFd, CallbackId, Response).
+
 %%% ============================================================================
 %%% Term to Python repr conversion
 %%% (Same as py_worker.erl - could be factored out to py_util.erl)

src/py_thread_handler.erl

@@ -22,7 +22,8 @@
     test_callback_with_complex_types/1,
     test_multiple_sequential_callbacks/1,
     test_call_from_non_worker_thread/1,
-    test_callback_with_try_except/1
+    test_callback_with_try_except/1,
+    test_async_call/1
 ]).
 
 all() ->
@@ -34,7 +35,8 @@ all() ->
         test_callback_with_complex_types,
         test_multiple_sequential_callbacks,
         test_call_from_non_worker_thread,
-        test_callback_with_try_except
+        test_callback_with_try_except,
+        test_async_call
     ].
 
 init_per_suite(Config) ->
@@ -58,6 +60,7 @@ end_per_testcase(_TestCase, _Config) ->
     catch py:unregister_function(add_ten),
     catch py:unregister_function(multiply_by_two),
     catch py:unregister_function(subtract_five),
+    catch py:unregister_function(async_multiply),
     ok.
 
 %%% ============================================================================
@@ -308,3 +311,27 @@ test_callback_with_try_except(_Config) ->
     %% Cleanup
     py:unregister_function(get_value),
     ok.
+
+%% @doc Test erlang.async_call() for asyncio-compatible callbacks.
+%% This tests the new async API that doesn't raise exceptions for control flow,
+%% making it safe to use from ASGI/asyncio applications.
+test_async_call(_Config) ->
+    %% Register an Erlang function
+    py:register_function(async_multiply, fun([X, Y]) -> X * Y end),
+
+    %% Add test directory to Python path
+    TestDir = code:lib_dir(erlang_python, test),
+    ok = py:exec(iolist_to_binary(io_lib:format(
+        "import sys; sys.path.insert(0, '~s')", [TestDir]))),
+
+    %% Test single async_call
+    {ok, Result} = py:call(py_test_async, run_single_test, []),
+    42 = Result,
+
+    %% Test concurrent async_calls
+    {ok, ConcurrentResult} = py:call(py_test_async, run_concurrent_test, []),
+    [6, 20, 42] = ConcurrentResult,
+
+    %% Cleanup
+    py:unregister_function(async_multiply),
+    ok.

test/py_reentrant_SUITE.erl

@@ -0,0 +1,34 @@
+"""Test module for erlang.async_call() functionality.
+
+This module tests the async API that integrates with asyncio without
+raising exceptions for control flow.
+"""
+
+import asyncio
+import erlang
+
+
+async def single_async_call():
+    """Test a single async_call."""
+    result = await erlang.async_call('async_multiply', 6, 7)
+    return result
+
+
+async def concurrent_async_calls():
+    """Test multiple concurrent async_calls using asyncio.gather."""
+    results = await asyncio.gather(
+        erlang.async_call('async_multiply', 2, 3),
+        erlang.async_call('async_multiply', 4, 5),
+        erlang.async_call('async_multiply', 6, 7)
+    )
+    return results
+
+
+def run_single_test():
+    """Run single async_call test."""
+    return asyncio.run(single_async_call())
+
+
+def run_concurrent_test():
+    """Run concurrent async_calls test."""
+    return asyncio.run(concurrent_async_calls())