Current section
Files
Jump to
Current section
Files
c_src/py_callback.c
/*
* 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.
*/
/**
* @file py_callback.c
* @brief Erlang callback support and asyncio integration
* @author Benoit Chesneau
*
* @ingroup cb
*
* This module implements bidirectional calling between Python and Erlang,
* enabling Python code to invoke Erlang functions and await their results.
*
* @par Features
*
* - **erlang module**: Python module providing `erlang.call()` and `erlang.func()`
* - **Suspension/Resume**: Reentrant callbacks without blocking dirty schedulers
* - **Asyncio support**: Background event loop for async Python operations
*
* @par Suspension Mechanism
*
* When Python calls `erlang.call('func', args)`:
*
* ```
* ┌────────────┐ ┌─────────────┐ ┌──────────────┐
* │ Python │ raises │ Executor │ returns │ Erlang │
* │ Code │ ──────> │ Catches │ ──────> │ Callback │
* └────────────┘ Suspend │ Exception │ suspend └──────────────┘
* └─────────────┘ │ │
* │ │ result
* ┌─────────────┐ │ │
* │ Resume │ <──────────────┘
* │ Replay │
* └─────────────┘
* │
* v
* ┌────────────┐
* │ Continue │
* │ Python │
* └────────────┘
* ```
*
* @par Why Suspension?
*
* Without suspension, Python calling Erlang would block a dirty scheduler
* while waiting for the Erlang callback to complete. With suspension:
*
* 1. Dirty scheduler is released immediately
* 2. Erlang callback runs on normal scheduler
* 3. Result is stored, Python is replayed on dirty scheduler
*
* @par The 'erlang' Python Module
*
* Provides two calling syntaxes:
*
* ```python
* # Explicit call
* result = erlang.call('my_function', arg1, arg2)
*
* # Attribute-style call (via __getattr__)
* result = erlang.my_function(arg1, arg2)
* ```
*
* @par Thread Safety
*
* - Thread-local storage tracks current worker and suspended state
* - Async event loop runs in dedicated thread
* - Pending futures queue protected by mutex
*
* @note This file is included from py_nif.c (single compilation unit)
*/
/* ============================================================================
* Callback Name Registry
*
* Maintains a C-side registry of registered callback function names.
* This allows erlang_module_getattr to only return ErlangFunction wrappers
* for actually registered functions, preventing introspection issues with
* libraries like torch that probe module attributes.
* ============================================================================ */
/**
* @def CALLBACK_REGISTRY_BUCKETS
* @brief Number of hash buckets for the callback registry
*/
#define CALLBACK_REGISTRY_BUCKETS 64
/**
* @struct callback_name_entry_t
* @brief Entry in the callback name registry hash table
*/
typedef struct callback_name_entry {
char *name; /**< Callback name (owned) */
size_t name_len; /**< Length of name */
struct callback_name_entry *next; /**< Next entry in bucket chain */
} callback_name_entry_t;
/** @brief Hash table buckets for callback registry */
static callback_name_entry_t *g_callback_registry[CALLBACK_REGISTRY_BUCKETS] = {NULL};
/** @brief Mutex protecting the callback registry */
static pthread_mutex_t g_callback_registry_mutex = PTHREAD_MUTEX_INITIALIZER;
/**
* @brief Simple hash function for callback names
*/
static unsigned int callback_name_hash(const char *name, size_t len) {
unsigned int hash = 5381;
for (size_t i = 0; i < len; i++) {
hash = ((hash << 5) + hash) + (unsigned char)name[i];
}
return hash % CALLBACK_REGISTRY_BUCKETS;
}
/**
* @brief Check if a callback name is registered
*
* Thread-safe lookup in the callback registry.
*
* @param name Callback name to check
* @param len Length of name
* @return true if registered, false otherwise
*/
static bool is_callback_registered(const char *name, size_t len) {
unsigned int bucket = callback_name_hash(name, len);
bool found = false;
pthread_mutex_lock(&g_callback_registry_mutex);
callback_name_entry_t *entry = g_callback_registry[bucket];
while (entry != NULL) {
if (entry->name_len == len && memcmp(entry->name, name, len) == 0) {
found = true;
break;
}
entry = entry->next;
}
pthread_mutex_unlock(&g_callback_registry_mutex);
return found;
}
/**
* @brief Register a callback name
*
* Thread-safe addition to the callback registry.
*
* @param name Callback name to register
* @param len Length of name
* @return 0 on success, -1 on failure
*/
static int register_callback_name(const char *name, size_t len) {
/* Check if already registered */
if (is_callback_registered(name, len)) {
return 0; /* Already registered, success */
}
/* Allocate new entry */
callback_name_entry_t *entry = enif_alloc(sizeof(callback_name_entry_t));
if (entry == NULL) {
return -1;
}
entry->name = enif_alloc(len + 1);
if (entry->name == NULL) {
enif_free(entry);
return -1;
}
memcpy(entry->name, name, len);
entry->name[len] = '\0';
entry->name_len = len;
unsigned int bucket = callback_name_hash(name, len);
pthread_mutex_lock(&g_callback_registry_mutex);
entry->next = g_callback_registry[bucket];
g_callback_registry[bucket] = entry;
pthread_mutex_unlock(&g_callback_registry_mutex);
return 0;
}
/**
* @brief Unregister a callback name
*
* Thread-safe removal from the callback registry.
*
* @param name Callback name to unregister
* @param len Length of name
*/
static void unregister_callback_name(const char *name, size_t len) {
unsigned int bucket = callback_name_hash(name, len);
pthread_mutex_lock(&g_callback_registry_mutex);
callback_name_entry_t **pp = &g_callback_registry[bucket];
while (*pp != NULL) {
callback_name_entry_t *entry = *pp;
if (entry->name_len == len && memcmp(entry->name, name, len) == 0) {
*pp = entry->next;
enif_free(entry->name);
enif_free(entry);
break;
}
pp = &entry->next;
}
pthread_mutex_unlock(&g_callback_registry_mutex);
}
/**
* @brief Clean up the callback registry
*
* Frees all entries. Called during NIF unload.
*/
static void cleanup_callback_registry(void) {
pthread_mutex_lock(&g_callback_registry_mutex);
for (int i = 0; i < CALLBACK_REGISTRY_BUCKETS; i++) {
callback_name_entry_t *entry = g_callback_registry[i];
while (entry != NULL) {
callback_name_entry_t *next = entry->next;
enif_free(entry->name);
enif_free(entry);
entry = next;
}
g_callback_registry[i] = NULL;
}
pthread_mutex_unlock(&g_callback_registry_mutex);
}
/* ============================================================================
* Suspended state management
* ============================================================================ */
/**
* Source type for suspended state creation.
* Indicates whether the source is a request or an existing suspended state.
*/
typedef enum {
SUSPENDED_SOURCE_REQUEST, /* Source is py_request_t */
SUSPENDED_SOURCE_EXISTING /* Source is suspended_state_t */
} suspended_source_type_t;
/**
* Source union for suspended state creation.
* Contains pointers to either request or existing suspended state.
*/
typedef struct {
suspended_source_type_t type;
union {
py_request_t *req; /* For SUSPENDED_SOURCE_REQUEST */
suspended_state_t *existing; /* For SUSPENDED_SOURCE_EXISTING */
} data;
} suspended_source_t;
/**
* Internal cleanup helper for suspended state creation failure.
*/
static void cleanup_suspended_state_partial(suspended_state_t *state, PyObject *callback_args) {
if (state->orig_env != NULL) {
enif_free_env(state->orig_env);
}
if (state->callback_args != NULL) {
Py_DECREF(state->callback_args);
} else if (callback_args != NULL) {
Py_DECREF(callback_args);
}
if (state->callback_func_name != NULL) {
enif_free(state->callback_func_name);
}
enif_release_resource(state);
}
/**
* Create a suspended state resource from exception args.
* Args tuple format: (callback_id, func_name, args)
*
* This unified function handles both:
* - Creating from a request (initial suspension)
* - Creating from an existing suspended state (nested suspension during replay)
*
* @param env NIF environment
* @param exc_args Exception args tuple from erlang.call()
* @param source Source of original request data
* @return suspended_state_t* or NULL on error
*/
static suspended_state_t *create_suspended_state_ex(
ErlNifEnv *env, PyObject *exc_args, const suspended_source_t *source) {
(void)env; /* Only needed for future extensions */
if (!PyTuple_Check(exc_args) || PyTuple_Size(exc_args) != 3) {
return NULL;
}
PyObject *callback_id_obj = PyTuple_GetItem(exc_args, 0);
PyObject *func_name_obj = PyTuple_GetItem(exc_args, 1);
PyObject *callback_args = PyTuple_GetItem(exc_args, 2);
if (!PyLong_Check(callback_id_obj) || !PyUnicode_Check(func_name_obj)) {
return NULL;
}
/* Allocate the suspended state resource */
suspended_state_t *state = enif_alloc_resource(
SUSPENDED_STATE_RESOURCE_TYPE, sizeof(suspended_state_t));
if (state == NULL) {
return NULL;
}
/* Initialize the state */
memset(state, 0, sizeof(suspended_state_t));
/* Set worker based on source type */
if (source->type == SUSPENDED_SOURCE_REQUEST) {
state->worker = tl_current_worker;
} else {
state->worker = source->data.existing->worker;
}
state->callback_id = PyLong_AsUnsignedLongLong(callback_id_obj);
/* Copy callback function name */
Py_ssize_t len;
const char *func_name = PyUnicode_AsUTF8AndSize(func_name_obj, &len);
if (func_name == NULL) {
enif_release_resource(state);
return NULL;
}
state->callback_func_name = enif_alloc(len + 1);
if (state->callback_func_name == NULL) {
enif_release_resource(state);
return NULL;
}
memcpy(state->callback_func_name, func_name, len);
state->callback_func_name[len] = '\0';
state->callback_func_len = len;
/* Store reference to callback args */
Py_INCREF(callback_args);
state->callback_args = callback_args;
/* Get request type and timeout based on source */
int request_type;
unsigned long timeout_ms;
if (source->type == SUSPENDED_SOURCE_REQUEST) {
request_type = source->data.req->type;
timeout_ms = source->data.req->timeout_ms;
} else {
request_type = source->data.existing->request_type;
timeout_ms = source->data.existing->orig_timeout_ms;
}
state->request_type = request_type;
state->orig_timeout_ms = timeout_ms;
/* Create environment to hold copied terms */
state->orig_env = enif_alloc_env();
if (state->orig_env == NULL) {
cleanup_suspended_state_partial(state, NULL);
return NULL;
}
/* Copy request-specific data based on source type and request type */
if (request_type == PY_REQ_CALL) {
ErlNifBinary *src_module, *src_func;
ERL_NIF_TERM src_args, src_kwargs;
ErlNifEnv *src_env;
if (source->type == SUSPENDED_SOURCE_REQUEST) {
src_module = &source->data.req->module_bin;
src_func = &source->data.req->func_bin;
src_args = source->data.req->args_term;
src_kwargs = source->data.req->kwargs_term;
src_env = source->data.req->env;
} else {
src_module = &source->data.existing->orig_module;
src_func = &source->data.existing->orig_func;
src_args = source->data.existing->orig_args;
src_kwargs = source->data.existing->orig_kwargs;
src_env = source->data.existing->orig_env;
}
/* Copy module binary */
if (!enif_alloc_binary(src_module->size, &state->orig_module)) {
cleanup_suspended_state_partial(state, NULL);
return NULL;
}
memcpy(state->orig_module.data, src_module->data, src_module->size);
/* Copy function binary */
if (!enif_alloc_binary(src_func->size, &state->orig_func)) {
enif_release_binary(&state->orig_module);
cleanup_suspended_state_partial(state, NULL);
return NULL;
}
memcpy(state->orig_func.data, src_func->data, src_func->size);
/* Copy args and kwargs to our environment */
state->orig_args = enif_make_copy(state->orig_env, src_args);
state->orig_kwargs = enif_make_copy(state->orig_env, src_kwargs);
(void)src_env; /* Used implicitly by enif_make_copy */
} else if (request_type == PY_REQ_EVAL) {
ErlNifBinary *src_code;
ERL_NIF_TERM src_locals;
ErlNifEnv *src_env;
if (source->type == SUSPENDED_SOURCE_REQUEST) {
src_code = &source->data.req->code_bin;
src_locals = source->data.req->locals_term;
src_env = source->data.req->env;
} else {
src_code = &source->data.existing->orig_code;
src_locals = source->data.existing->orig_locals;
src_env = source->data.existing->orig_env;
}
/* Copy code binary */
if (!enif_alloc_binary(src_code->size, &state->orig_code)) {
cleanup_suspended_state_partial(state, NULL);
return NULL;
}
memcpy(state->orig_code.data, src_code->data, src_code->size);
/* Copy locals */
state->orig_locals = enif_make_copy(state->orig_env, src_locals);
(void)src_env; /* Used implicitly by enif_make_copy */
}
/* Initialize synchronization primitives */
pthread_mutex_init(&state->mutex, NULL);
pthread_cond_init(&state->cond, NULL);
state->result_data = NULL;
state->result_len = 0;
state->has_result = false;
state->is_error = false;
return state;
}
/**
* Create a suspended state resource from a request.
* Wrapper for create_suspended_state_ex for initial suspension.
*/
static suspended_state_t *create_suspended_state(ErlNifEnv *env, PyObject *exc_args,
py_request_t *req) {
suspended_source_t source = {
.type = SUSPENDED_SOURCE_REQUEST,
.data.req = req
};
return create_suspended_state_ex(env, exc_args, &source);
}
/**
* Create a new suspended state from an existing one (for nested suspensions).
* Wrapper for create_suspended_state_ex for nested suspension during replay.
*/
static suspended_state_t *create_suspended_state_from_existing(
ErlNifEnv *env, PyObject *exc_args, suspended_state_t *existing) {
suspended_source_t source = {
.type = SUSPENDED_SOURCE_EXISTING,
.data.existing = existing
};
return create_suspended_state_ex(env, exc_args, &source);
}
/**
* Build exception args tuple from thread-local pending callback state.
*
* This helper extracts the common pattern of building the exc_args tuple
* (callback_id, func_name, args) from thread-local storage.
*
* @return PyObject* tuple on success, NULL on failure
* @note On failure, tl_pending_callback is cleared
* @note Caller must Py_DECREF the returned tuple when done
*/
static PyObject *build_pending_callback_exc_args(void) {
PyObject *exc_args = PyTuple_New(3);
if (exc_args == NULL) {
tl_pending_callback = false;
return NULL;
}
PyObject *callback_id_obj = PyLong_FromUnsignedLongLong(tl_pending_callback_id);
PyObject *func_name_obj = PyUnicode_FromStringAndSize(
tl_pending_func_name, tl_pending_func_name_len);
if (callback_id_obj == NULL || func_name_obj == NULL) {
Py_XDECREF(callback_id_obj);
Py_XDECREF(func_name_obj);
Py_DECREF(exc_args);
tl_pending_callback = false;
return NULL;
}
PyTuple_SET_ITEM(exc_args, 0, callback_id_obj);
PyTuple_SET_ITEM(exc_args, 1, func_name_obj);
Py_INCREF(tl_pending_args); /* Tuple takes ownership */
PyTuple_SET_ITEM(exc_args, 2, tl_pending_args);
return exc_args;
}
/**
* Build the {suspended, ...} result term from a suspended state.
*
* Common helper for creating the suspension result after a callback
* is detected during Python execution.
*
* @param env NIF environment
* @param suspended Suspended state (resource will be released)
* @return ERL_NIF_TERM {suspended, CallbackId, StateRef, {FuncName, Args}}
* @note Clears tl_pending_callback
*/
static ERL_NIF_TERM build_suspended_result(ErlNifEnv *env, suspended_state_t *suspended) {
ERL_NIF_TERM state_ref = enif_make_resource(env, suspended);
enif_release_resource(suspended);
ERL_NIF_TERM callback_id_term = enif_make_uint64(env, tl_pending_callback_id);
ERL_NIF_TERM func_name_term;
unsigned char *fn_buf = enif_make_new_binary(env, tl_pending_func_name_len, &func_name_term);
memcpy(fn_buf, tl_pending_func_name, tl_pending_func_name_len);
ERL_NIF_TERM args_term = py_to_term(env, tl_pending_args);
tl_pending_callback = false;
return enif_make_tuple4(env,
ATOM_SUSPENDED,
callback_id_term,
state_ref,
enif_make_tuple2(env, func_name_term, args_term));
}
/**
* Helper to parse callback response data into a Python object.
* Response format: status_byte (0=ok, 1=error) + python_repr_string
*/
static PyObject *parse_callback_response(unsigned char *response_data, size_t response_len) {
if (response_len < 1) {
PyErr_SetString(PyExc_RuntimeError, "Empty callback response");
return NULL;
}
uint8_t status = response_data[0];
if (response_len < 2) {
if (status == 0) {
Py_RETURN_NONE;
} else {
PyErr_SetString(PyExc_RuntimeError, "Erlang callback failed");
return NULL;
}
}
char *result_str = (char *)response_data + 1;
size_t result_len = response_len - 1;
PyObject *result = NULL;
if (status == 0) {
/* Try to evaluate the result string as Python literal */
PyObject *ast_module = PyImport_ImportModule("ast");
if (ast_module != NULL) {
PyObject *literal_eval = PyObject_GetAttrString(ast_module, "literal_eval");
if (literal_eval != NULL) {
PyObject *arg = PyUnicode_FromStringAndSize(result_str, result_len);
if (arg != NULL) {
result = PyObject_CallFunctionObjArgs(literal_eval, arg, NULL);
Py_DECREF(arg);
if (result == NULL) {
/* If literal_eval fails, return as string */
PyErr_Clear();
result = PyUnicode_FromStringAndSize(result_str, result_len);
}
}
Py_DECREF(literal_eval);
}
Py_DECREF(ast_module);
}
if (result == NULL) {
result = PyUnicode_FromStringAndSize(result_str, result_len);
}
} else {
/* Error case */
char *err_msg = enif_alloc(result_len + 1);
if (err_msg != NULL) {
memcpy(err_msg, result_str, result_len);
err_msg[result_len] = '\0';
PyErr_SetString(PyExc_RuntimeError, err_msg);
enif_free(err_msg);
} else {
PyErr_SetString(PyExc_RuntimeError, "Erlang callback failed");
}
}
return result;
}
/* ============================================================================
* Erlang callback module for Python
* ============================================================================ */
/* ErlangFunction - callable wrapper for registered Erlang functions */
typedef struct {
PyObject_HEAD
PyObject *name; /* Function name as Python string */
} ErlangFunctionObject;
static void ErlangFunction_dealloc(ErlangFunctionObject *self) {
Py_XDECREF(self->name);
Py_TYPE(self)->tp_free((PyObject *)self);
}
/* Forward declaration - implemented after erlang_call_impl */
static PyObject *ErlangFunction_call(ErlangFunctionObject *self, PyObject *args, PyObject *kwds);
static PyObject *ErlangFunction_repr(ErlangFunctionObject *self) {
return PyUnicode_FromFormat("<erlang function '%U'>", self->name);
}
static PyTypeObject ErlangFunctionType = {
PyVarObject_HEAD_INIT(NULL, 0)
.tp_name = "erlang.Function",
.tp_doc = "Wrapper for registered Erlang function",
.tp_basicsize = sizeof(ErlangFunctionObject),
.tp_itemsize = 0,
.tp_flags = Py_TPFLAGS_DEFAULT,
.tp_dealloc = (destructor)ErlangFunction_dealloc,
.tp_call = (ternaryfunc)ErlangFunction_call,
.tp_repr = (reprfunc)ErlangFunction_repr,
};
/* Helper to create ErlangFunction instance */
static PyObject *ErlangFunction_New(PyObject *name) {
ErlangFunctionObject *self = PyObject_New(ErlangFunctionObject, &ErlangFunctionType);
if (self != NULL) {
Py_INCREF(name);
self->name = name;
}
return (PyObject *)self;
}
/**
* Python implementation of erlang.call(name, *args)
*
* This function allows Python code to call registered Erlang functions.
*
* The implementation uses a suspension/resume mechanism to avoid holding
* dirty schedulers during callbacks:
*
* 1. If a suspended state exists with a cached result, return it immediately
* 2. Otherwise, create a suspended state, send callback message, wait on condvar
* 3. When resume_callback is called, the condvar is signaled with the result
* 4. Parse and return the result
*
* This allows the dirty scheduler to be freed while waiting for the callback.
*/
static PyObject *erlang_call_impl(PyObject *self, PyObject *args) {
(void)self;
/*
* Check if this is a call from an executor thread (normal path) or
* from a spawned thread (thread worker path).
*/
if (tl_current_worker == NULL || !tl_current_worker->has_callback_handler) {
/*
* Not an executor thread - use thread worker path.
* This enables any spawned Python thread to call erlang.call():
* - threading.Thread instances
* - concurrent.futures.ThreadPoolExecutor workers
* - Any other Python threads
*/
Py_ssize_t nargs = PyTuple_Size(args);
if (nargs < 1) {
PyErr_SetString(PyExc_TypeError, "erlang.call requires at least a function name");
return NULL;
}
PyObject *name_obj = PyTuple_GetItem(args, 0);
if (!PyUnicode_Check(name_obj)) {
PyErr_SetString(PyExc_TypeError, "Function name must be a string");
return NULL;
}
const char *func_name = PyUnicode_AsUTF8(name_obj);
if (func_name == NULL) {
return NULL;
}
size_t func_name_len = strlen(func_name);
/* Build args list (remaining args) */
PyObject *call_args = PyTuple_GetSlice(args, 1, nargs);
if (call_args == NULL) {
return NULL;
}
/* Use thread worker call */
PyObject *result = thread_worker_call(func_name, func_name_len, call_args);
Py_DECREF(call_args);
return result;
}
Py_ssize_t nargs = PyTuple_Size(args);
if (nargs < 1) {
PyErr_SetString(PyExc_TypeError, "erlang.call requires at least a function name");
return NULL;
}
/* Get function name (first arg) */
PyObject *name_obj = PyTuple_GetItem(args, 0);
if (!PyUnicode_Check(name_obj)) {
PyErr_SetString(PyExc_TypeError, "Function name must be a string");
return NULL;
}
const char *func_name = PyUnicode_AsUTF8(name_obj);
if (func_name == NULL) {
return NULL;
}
size_t func_name_len = strlen(func_name);
/* Check if we have a suspended state with a cached result (replay case) */
if (tl_current_suspended != NULL && tl_current_suspended->has_result) {
/* Verify this is the same callback */
if (tl_current_suspended->callback_func_len == func_name_len &&
memcmp(tl_current_suspended->callback_func_name, func_name, func_name_len) == 0) {
/* Return the cached result - parse using ast.literal_eval */
PyObject *result = parse_callback_response(
tl_current_suspended->result_data,
tl_current_suspended->result_len);
/* Mark result as consumed (don't clear tl_current_suspended yet,
* as we might need it for nested callbacks in the future) */
tl_current_suspended->has_result = false;
return result;
}
}
/*
* FIX for multiple sequential erlang.call():
* If we're in replay context (tl_current_suspended != NULL) but didn't get
* a cache hit above, this is a SUBSEQUENT call (e.g., second erlang.call()
* in the same Python function). We MUST NOT suspend again - that would
* cause an infinite loop where replay always hits this second call.
* Instead, fall through to blocking pipe behavior for subsequent calls.
*/
bool force_blocking = (tl_current_suspended != NULL);
/* Build args list (remaining args) */
PyObject *call_args = PyTuple_GetSlice(args, 1, nargs);
if (call_args == NULL) {
return NULL;
}
/*
* Check if suspension is allowed.
* Suspension is only safe when the result will be directly examined by the
* executor (PY_REQ_CALL or PY_REQ_EVAL). For PY_REQ_EXEC or nested Python
* code, we must block and wait for the result.
*
* Also block if force_blocking is set (replay context with no cache hit).
*/
if (!tl_allow_suspension || force_blocking) {
/* Fall back to blocking behavior - send message and wait on pipe */
ErlNifEnv *msg_env = enif_alloc_env();
if (msg_env == NULL) {
Py_DECREF(call_args);
PyErr_SetString(PyExc_MemoryError, "Failed to allocate message environment");
return NULL;
}
ERL_NIF_TERM func_term;
{
unsigned char *buf = enif_make_new_binary(msg_env, func_name_len, &func_term);
memcpy(buf, func_name, func_name_len);
}
ERL_NIF_TERM args_term = py_to_term(msg_env, call_args);
Py_DECREF(call_args);
uint64_t callback_id = atomic_fetch_add(&g_callback_id_counter, 1);
ERL_NIF_TERM id_term = enif_make_uint64(msg_env, callback_id);
ERL_NIF_TERM msg = enif_make_tuple4(msg_env,
ATOM_ERLANG_CALLBACK,
id_term,
func_term,
args_term);
char *response_data = NULL;
uint32_t response_len = 0;
int read_result;
Py_BEGIN_ALLOW_THREADS
enif_send(NULL, &tl_current_worker->callback_handler, msg_env, msg);
enif_free_env(msg_env);
/* Use 30 second timeout to prevent indefinite blocking */
read_result = read_length_prefixed_data(
tl_current_worker->callback_pipe[0],
&response_data, &response_len, 30000);
Py_END_ALLOW_THREADS
if (read_result == -1) {
if (errno == ETIMEDOUT) {
PyErr_SetString(PyExc_TimeoutError, "Callback response timed out");
} else {
PyErr_SetString(PyExc_RuntimeError, "Failed to read callback response");
}
return NULL;
}
if (read_result == -2) {
PyErr_SetString(PyExc_MemoryError, "Failed to allocate response buffer");
return NULL;
}
PyObject *result = parse_callback_response((unsigned char *)response_data, response_len);
if (response_data != NULL) {
enif_free(response_data);
}
return result;
}
/*
* Flag-based suspension: set thread-local flag and raise exception.
*
* Unlike checking exception type (which fails if frameworks catch exceptions),
* we set a thread-local flag that the C executor checks FIRST. This way:
* 1. Python code can catch/re-raise the exception - we don't care
* 2. The flag tells us a callback is pending
* 3. Executor handles it before looking at exception type
*
* The exception is just to abort Python execution cleanly.
*/
uint64_t callback_id = atomic_fetch_add(&g_callback_id_counter, 1);
/* Set pending callback flag and store info */
tl_pending_callback = true;
tl_pending_callback_id = callback_id;
/* Store function name (make a copy) */
if (tl_pending_func_name != NULL) {
enif_free(tl_pending_func_name);
}
tl_pending_func_name = enif_alloc(func_name_len + 1);
if (tl_pending_func_name == NULL) {
tl_pending_callback = false;
Py_DECREF(call_args);
PyErr_SetString(PyExc_MemoryError, "Failed to allocate function name");
return NULL;
}
memcpy(tl_pending_func_name, func_name, func_name_len);
tl_pending_func_name[func_name_len] = '\0';
tl_pending_func_name_len = func_name_len;
/* Store args (take ownership) */
Py_XDECREF(tl_pending_args);
tl_pending_args = call_args; /* Takes ownership, don't decref */
/* Raise exception to abort Python execution */
PyErr_SetString(SuspensionRequiredException, "callback pending");
return NULL;
}
/* ============================================================================
* Async callback support for asyncio integration
*
* This provides erlang.async_call() which returns an asyncio.Future that
* resolves when the Erlang callback completes. Unlike erlang.call():
* - No exceptions raised for control flow
* - Integrates with asyncio event loop
* - Releases dirty NIF thread while waiting
* ============================================================================ */
/*
* Forward declarations for thread worker variables (defined in py_thread_worker.c)
* These are needed because py_callback.c is included before py_thread_worker.c.
*/
extern ErlNifPid g_thread_coordinator_pid;
extern bool g_has_thread_coordinator;
/* Global state for async callbacks */
static int g_async_callback_pipe[2] = {-1, -1}; /* [0]=read, [1]=write */
static PyObject *g_async_pending_futures = NULL; /* Dict: callback_id -> Future */
static pthread_mutex_t g_async_futures_mutex = PTHREAD_MUTEX_INITIALIZER;
/* Thread-safe initialization using pthread_once */
static pthread_once_t g_async_callback_init_once = PTHREAD_ONCE_INIT;
static int g_async_callback_init_result = 0;
/**
* Internal initialization function called by pthread_once.
* Thread-safe: only called once by pthread_once.
*/
static void async_callback_init_impl(void) {
if (pipe(g_async_callback_pipe) < 0) {
g_async_callback_init_result = -1;
return;
}
/* Set the read end to non-blocking for asyncio compatibility */
int flags = fcntl(g_async_callback_pipe[0], F_GETFL, 0);
if (flags >= 0) {
fcntl(g_async_callback_pipe[0], F_SETFL, flags | O_NONBLOCK);
}
g_async_pending_futures = PyDict_New();
if (g_async_pending_futures == NULL) {
close(g_async_callback_pipe[0]);
close(g_async_callback_pipe[1]);
g_async_callback_pipe[0] = -1;
g_async_callback_pipe[1] = -1;
g_async_callback_init_result = -1;
return;
}
g_async_callback_init_result = 0;
}
/**
* Initialize async callback system.
* Creates the response pipe and pending futures dict.
* Thread-safe: uses pthread_once for initialization.
*/
static int async_callback_init(void) {
pthread_once(&g_async_callback_init_once, async_callback_init_impl);
return g_async_callback_init_result;
}
/**
* Process a single async callback response from the pipe.
* Called by the asyncio reader callback.
* Returns: 1 if processed, 0 if no data, -1 on error
*/
static int process_async_callback_response(void) {
/* Read callback_id (8 bytes) + response_len (4 bytes) + response_data */
uint64_t callback_id;
uint32_t response_len;
ssize_t n;
n = read(g_async_callback_pipe[0], &callback_id, sizeof(callback_id));
if (n < 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
return 0; /* No data available (non-blocking) */
}
return -1; /* Error */
}
if (n == 0) {
return 0; /* EOF / No data */
}
if (n != sizeof(callback_id)) {
return -1; /* Partial read - error */
}
n = read(g_async_callback_pipe[0], &response_len, sizeof(response_len));
if (n != sizeof(response_len)) {
return -1;
}
char *response_data = NULL;
if (response_len > 0) {
response_data = enif_alloc(response_len);
if (response_data == NULL) {
return -1;
}
n = read(g_async_callback_pipe[0], response_data, response_len);
if (n != (ssize_t)response_len) {
enif_free(response_data);
return -1;
}
}
/* Look up and resolve the Future */
pthread_mutex_lock(&g_async_futures_mutex);
PyObject *key = PyLong_FromUnsignedLongLong(callback_id);
PyObject *future = PyDict_GetItem(g_async_pending_futures, key);
if (future != NULL) {
Py_INCREF(future); /* Keep reference while we use it */
PyDict_DelItem(g_async_pending_futures, key);
}
Py_DECREF(key);
pthread_mutex_unlock(&g_async_futures_mutex);
if (future != NULL) {
/* Parse response and resolve Future */
PyObject *result = NULL;
if (response_data != NULL) {
result = parse_callback_response((unsigned char *)response_data, response_len);
} else {
Py_INCREF(Py_None);
result = Py_None;
}
if (result != NULL) {
/* Call future.set_result(result) */
PyObject *set_result = PyObject_GetAttrString(future, "set_result");
if (set_result != NULL) {
PyObject *ret = PyObject_CallFunctionObjArgs(set_result, result, NULL);
Py_XDECREF(ret);
Py_DECREF(set_result);
}
Py_DECREF(result);
} else {
/* Error occurred - set exception on Future */
PyObject *exc_type, *exc_value, *exc_tb;
PyErr_Fetch(&exc_type, &exc_value, &exc_tb);
PyObject *set_exception = PyObject_GetAttrString(future, "set_exception");
if (set_exception != NULL) {
if (exc_value != NULL) {
PyObject *ret = PyObject_CallFunctionObjArgs(set_exception, exc_value, NULL);
Py_XDECREF(ret);
} else {
PyObject *runtime_err = PyObject_CallFunction(PyExc_RuntimeError,
"s", "Erlang callback failed");
PyObject *ret = PyObject_CallFunctionObjArgs(set_exception, runtime_err, NULL);
Py_XDECREF(ret);
Py_XDECREF(runtime_err);
}
Py_DECREF(set_exception);
}
Py_XDECREF(exc_type);
Py_XDECREF(exc_value);
Py_XDECREF(exc_tb);
PyErr_Clear();
}
Py_DECREF(future);
}
if (response_data != NULL) {
enif_free(response_data);
}
return 1;
}
/**
* Python callback for asyncio reader.
* Called when data is available on the async callback pipe.
*/
static PyObject *async_callback_reader(PyObject *self, PyObject *args) {
(void)self;
(void)args;
/* Process all available responses */
while (process_async_callback_response() > 0) {
/* Continue processing */
}
Py_RETURN_NONE;
}
/**
* Get the read file descriptor for the async callback pipe.
* Used by Python to register with asyncio.
*/
static PyObject *get_async_callback_fd(PyObject *self, PyObject *args) {
(void)self;
(void)args;
/* async_callback_init uses pthread_once, so it's safe to call multiple times */
if (async_callback_init() < 0) {
PyErr_SetString(PyExc_RuntimeError, "Failed to initialize async callback system");
return NULL;
}
return PyLong_FromLong(g_async_callback_pipe[0]);
}
/**
* Send an async callback request to Erlang.
* Returns the callback_id for tracking.
*/
static PyObject *send_async_callback_request(PyObject *self, PyObject *args) {
(void)self;
PyObject *name_obj;
PyObject *call_args;
if (!PyArg_ParseTuple(args, "OO", &name_obj, &call_args)) {
return NULL;
}
if (!PyUnicode_Check(name_obj)) {
PyErr_SetString(PyExc_TypeError, "Function name must be a string");
return NULL;
}
if (!PyTuple_Check(call_args)) {
PyErr_SetString(PyExc_TypeError, "Arguments must be a tuple");
return NULL;
}
const char *func_name = PyUnicode_AsUTF8(name_obj);
if (func_name == NULL) {
return NULL;
}
size_t func_name_len = strlen(func_name);
/* Check if thread worker coordinator is available */
if (!g_has_thread_coordinator) {
PyErr_SetString(PyExc_RuntimeError,
"Thread worker coordinator not initialized. "
"Ensure erlang_python application is started.");
return NULL;
}
/* Generate callback ID */
uint64_t callback_id = atomic_fetch_add(&g_callback_id_counter, 1);
/* Send callback request to Erlang via thread worker coordinator */
ErlNifEnv *msg_env = enif_alloc_env();
if (msg_env == NULL) {
PyErr_SetString(PyExc_MemoryError, "Failed to allocate message environment");
return NULL;
}
/* Create function name binary */
ERL_NIF_TERM func_term;
unsigned char *fn_buf = enif_make_new_binary(msg_env, func_name_len, &func_term);
if (fn_buf == NULL) {
enif_free_env(msg_env);
PyErr_SetString(PyExc_MemoryError, "Failed to allocate function name");
return NULL;
}
memcpy(fn_buf, func_name, func_name_len);
/* Convert args to Erlang term */
ERL_NIF_TERM args_term = py_to_term(msg_env, call_args);
ERL_NIF_TERM id_term = enif_make_uint64(msg_env, callback_id);
/* Send message: {async_callback, CallbackId, FuncName, Args, WriteFd}
* The WriteFd is the async callback pipe write end */
ERL_NIF_TERM msg = enif_make_tuple5(msg_env,
enif_make_atom(msg_env, "async_callback"),
id_term,
func_term,
args_term,
enif_make_int(msg_env, g_async_callback_pipe[1]));
if (!enif_send(NULL, &g_thread_coordinator_pid, msg_env, msg)) {
enif_free_env(msg_env);
PyErr_SetString(PyExc_RuntimeError, "Failed to send async callback message");
return NULL;
}
enif_free_env(msg_env);
return PyLong_FromUnsignedLongLong(callback_id);
}
/**
* Register a Future for an async callback.
*/
static PyObject *register_async_future(PyObject *self, PyObject *args) {
(void)self;
unsigned long long callback_id;
PyObject *future;
if (!PyArg_ParseTuple(args, "KO", &callback_id, &future)) {
return NULL;
}
pthread_mutex_lock(&g_async_futures_mutex);
PyObject *key = PyLong_FromUnsignedLongLong(callback_id);
Py_INCREF(future);
PyDict_SetItem(g_async_pending_futures, key, future);
Py_DECREF(key);
pthread_mutex_unlock(&g_async_futures_mutex);
Py_RETURN_NONE;
}
/**
* ErlangFunction.__call__ - forward to erlang_call_impl
*/
static PyObject *ErlangFunction_call(ErlangFunctionObject *self, PyObject *args, PyObject *kwds) {
(void)kwds; /* Unused */
/* Build new args tuple: (name, *args) */
Py_ssize_t nargs = PyTuple_Size(args);
PyObject *new_args = PyTuple_New(nargs + 1);
if (new_args == NULL) {
return NULL;
}
Py_INCREF(self->name);
PyTuple_SET_ITEM(new_args, 0, self->name);
for (Py_ssize_t i = 0; i < nargs; i++) {
PyObject *item = PyTuple_GET_ITEM(args, i);
Py_INCREF(item);
PyTuple_SET_ITEM(new_args, i + 1, item);
}
/* Call existing erlang_call_impl */
PyObject *result = erlang_call_impl(NULL, new_args);
Py_DECREF(new_args);
return result;
}
/**
* Module __getattr__ - enables "from erlang import func_name" and "erlang.func_name()"
*
* Only returns ErlangFunction wrapper for REGISTERED callback names.
* This prevents torch and other libraries that introspect module attributes
* from getting callable objects for arbitrary attribute names.
*/
static PyObject *erlang_module_getattr(PyObject *module, PyObject *name) {
(void)module; /* Unused */
/* Get the name as a C string */
const char *name_str = PyUnicode_AsUTF8(name);
if (name_str == NULL) {
return NULL; /* Exception already set */
}
size_t name_len = strlen(name_str);
/* Check if this callback is registered */
if (!is_callback_registered(name_str, name_len)) {
PyErr_Format(PyExc_AttributeError,
"module 'erlang' has no attribute '%s'", name_str);
return NULL;
}
/* Return an ErlangFunction wrapper for registered callbacks */
return ErlangFunction_New(name);
}
/* Python method definitions for erlang module */
static PyMethodDef ErlangModuleMethods[] = {
{"call", erlang_call_impl, METH_VARARGS,
"Call a registered Erlang function.\n\n"
"Usage: erlang.call('func_name', arg1, arg2, ...)\n"
"Returns: The result from the Erlang function."},
{"_get_async_callback_fd", get_async_callback_fd, METH_NOARGS,
"Get the file descriptor for async callback responses.\n"
"Used internally by async_call() to register with asyncio."},
{"_async_callback_reader", async_callback_reader, METH_NOARGS,
"Process pending async callback responses.\n"
"Called by asyncio when the callback pipe has data."},
{"_send_async_request", send_async_callback_request, METH_VARARGS,
"Send an async callback request to Erlang.\n"
"Returns the callback_id for tracking."},
{"_register_async_future", register_async_future, METH_VARARGS,
"Register a Future for an async callback.\n"
"Usage: erlang._register_async_future(callback_id, future)"},
{NULL, NULL, 0, NULL}
};
/* Module __getattr__ method definition (for adding to module dict) */
static PyMethodDef getattr_method = {
"__getattr__", erlang_module_getattr, METH_O,
"Get an Erlang function wrapper by name."
};
/* Module definition */
static struct PyModuleDef ErlangModuleDef = {
PyModuleDef_HEAD_INIT,
"erlang", /* Module name */
"Interface for calling Erlang functions from Python.", /* Docstring */
-1, /* Size of per-interpreter state (-1 = global) */
ErlangModuleMethods /* Methods */
};
/**
* Create and register the 'erlang' module in Python.
* Called during Python initialization.
*/
static int create_erlang_module(void) {
/* Initialize ErlangFunction type */
if (PyType_Ready(&ErlangFunctionType) < 0) {
return -1;
}
PyObject *module = PyModule_Create(&ErlangModuleDef);
if (module == NULL) {
return -1;
}
/* Create the SuspensionRequired exception.
* This exception is raised internally when erlang.call() needs to suspend.
* It carries callback info in args: (callback_id, func_name, args_tuple) */
SuspensionRequiredException = PyErr_NewException(
"erlang.SuspensionRequired", NULL, NULL);
if (SuspensionRequiredException == NULL) {
Py_DECREF(module);
return -1;
}
Py_INCREF(SuspensionRequiredException);
if (PyModule_AddObject(module, "SuspensionRequired", SuspensionRequiredException) < 0) {
Py_DECREF(SuspensionRequiredException);
Py_DECREF(module);
return -1;
}
/* Add ErlangFunction type to module (for introspection) */
Py_INCREF(&ErlangFunctionType);
if (PyModule_AddObject(module, "Function", (PyObject *)&ErlangFunctionType) < 0) {
Py_DECREF(&ErlangFunctionType);
Py_DECREF(module);
return -1;
}
/* Add __getattr__ to enable "from erlang import name" and "erlang.name()" syntax
* Module __getattr__ (PEP 562) needs to be set as an attribute on the module dict */
PyObject *getattr_func = PyCFunction_New(&getattr_method, module);
if (getattr_func == NULL) {
Py_DECREF(module);
return -1;
}
if (PyModule_AddObject(module, "__getattr__", getattr_func) < 0) {
Py_DECREF(getattr_func);
Py_DECREF(module);
return -1;
}
/* Add module to sys.modules */
PyObject *sys_modules = PyImport_GetModuleDict();
if (PyDict_SetItemString(sys_modules, "erlang", module) < 0) {
Py_DECREF(module);
return -1;
}
/* Add the async_call() coroutine function.
* This is implemented in Python for easier asyncio integration. */
const char *async_call_code =
"import asyncio\n"
"import erlang\n"
"\n"
"# Track if we've registered the reader with the event loop\n"
"_async_reader_registered = {}\n"
"\n"
"async def async_call(func_name, *args):\n"
" '''\n"
" Call an Erlang function asynchronously.\n"
" \n"
" This is safe to use from asyncio code:\n"
" - No exceptions raised for control flow\n"
" - Integrates with asyncio event loop\n"
" - Releases dirty NIF thread while waiting\n"
" \n"
" Usage:\n"
" result = await erlang.async_call('my_function', arg1, arg2)\n"
" \n"
" Args:\n"
" func_name: Name of the registered Erlang function\n"
" *args: Arguments to pass to the function\n"
" \n"
" Returns:\n"
" The result from the Erlang function\n"
" '''\n"
" loop = asyncio.get_running_loop()\n"
" \n"
" # Ensure the reader is registered with this event loop\n"
" loop_id = id(loop)\n"
" if loop_id not in _async_reader_registered:\n"
" fd = erlang._get_async_callback_fd()\n"
" loop.add_reader(fd, erlang._async_callback_reader)\n"
" _async_reader_registered[loop_id] = True\n"
" \n"
" # Create a Future for this call\n"
" future = loop.create_future()\n"
" \n"
" # Send the request and get callback_id\n"
" callback_id = erlang._send_async_request(func_name, args)\n"
" \n"
" # Register the Future\n"
" erlang._register_async_future(callback_id, future)\n"
" \n"
" # Wait for the result\n"
" return await future\n"
"\n"
"# Add async_call to the erlang module\n"
"erlang.async_call = async_call\n"
"erlang._async_reader_registered = _async_reader_registered\n";
PyObject *globals = PyDict_New();
if (globals == NULL) {
/* Non-fatal - async_call just won't be available */
PyErr_Clear();
} else {
PyObject *builtins = PyEval_GetBuiltins();
PyDict_SetItemString(globals, "__builtins__", builtins);
PyObject *result = PyRun_String(async_call_code, Py_file_input, globals, globals);
if (result == NULL) {
/* Non-fatal - async_call just won't be available */
PyErr_Print();
PyErr_Clear();
} else {
Py_DECREF(result);
}
Py_DECREF(globals);
}
return 0;
}
/* ============================================================================
* Asyncio support
* ============================================================================ */
/**
* Callback function that gets invoked when a future completes.
* This is called from within the event loop thread.
*/
static void async_future_callback(py_async_worker_t *worker, async_pending_t *pending) {
ErlNifEnv *msg_env = enif_alloc_env();
if (msg_env == NULL) {
/* Cannot send result - just log and return */
return;
}
PyObject *py_result = PyObject_CallMethod(pending->future, "result", NULL);
ERL_NIF_TERM result_term;
if (py_result == NULL) {
/* Exception occurred */
PyObject *exc = PyObject_CallMethod(pending->future, "exception", NULL);
if (exc != NULL && exc != Py_None) {
PyObject *str = PyObject_Str(exc);
const char *err_msg = str ? PyUnicode_AsUTF8(str) : "unknown";
result_term = enif_make_tuple2(msg_env, ATOM_ERROR,
enif_make_string(msg_env, err_msg, ERL_NIF_LATIN1));
Py_XDECREF(str);
} else {
result_term = enif_make_tuple2(msg_env, ATOM_ERROR,
enif_make_atom(msg_env, "unknown"));
}
Py_XDECREF(exc);
PyErr_Clear();
} else {
result_term = enif_make_tuple2(msg_env, ATOM_OK,
py_to_term(msg_env, py_result));
Py_DECREF(py_result);
}
/* Send message: {async_result, Id, Result} */
ERL_NIF_TERM msg = enif_make_tuple3(msg_env,
ATOM_ASYNC_RESULT,
enif_make_uint64(msg_env, pending->id),
result_term);
enif_send(NULL, &pending->caller, msg_env, msg);
enif_free_env(msg_env);
}
/**
* Background thread running the asyncio event loop.
* This thread owns the event loop and processes coroutines.
*/
static void *async_event_loop_thread(void *arg) {
py_async_worker_t *worker = (py_async_worker_t *)arg;
/* Acquire GIL for this thread */
PyGILState_STATE gstate = PyGILState_Ensure();
/* Import asyncio */
PyObject *asyncio = PyImport_ImportModule("asyncio");
if (asyncio == NULL) {
PyErr_Print();
PyGILState_Release(gstate);
worker->loop_running = false;
return NULL;
}
/* Create new event loop */
PyObject *loop = PyObject_CallMethod(asyncio, "new_event_loop", NULL);
if (loop == NULL) {
PyErr_Print();
Py_DECREF(asyncio);
PyGILState_Release(gstate);
worker->loop_running = false;
return NULL;
}
/* Set as current loop */
PyObject *set_result = PyObject_CallMethod(asyncio, "set_event_loop", "O", loop);
Py_XDECREF(set_result);
worker->event_loop = loop;
Py_INCREF(loop); /* Keep extra ref for worker struct */
Py_DECREF(asyncio);
worker->loop_running = true;
/* Run the event loop with proper GIL management */
while (!worker->shutdown) {
/* Release GIL while sleeping (allow other Python threads to run) */
Py_BEGIN_ALLOW_THREADS
usleep(10000); /* 10ms sleep without holding GIL */
Py_END_ALLOW_THREADS
/* Run one iteration of the event loop with GIL held */
PyObject *asyncio_mod = PyImport_ImportModule("asyncio");
if (asyncio_mod != NULL) {
PyObject *sleep_coro = PyObject_CallMethod(asyncio_mod, "sleep", "d", 0.0);
if (sleep_coro != NULL) {
PyObject *task = PyObject_CallMethod(loop, "create_task", "O", sleep_coro);
Py_DECREF(sleep_coro);
if (task != NULL) {
PyObject *run_result = PyObject_CallMethod(loop, "run_until_complete", "O", task);
Py_DECREF(task);
Py_XDECREF(run_result);
}
}
Py_DECREF(asyncio_mod);
}
if (PyErr_Occurred()) {
PyErr_Clear();
}
/*
* Check for completed futures (GIL held).
*
* IMPORTANT: We must not hold the mutex while calling Python functions
* to avoid deadlocks. The pattern is:
* 1. Lock mutex, collect completed items, unlock
* 2. Process callbacks outside mutex (no contention)
* 3. Lock mutex, remove processed items, unlock
*/
/* Phase 1: Collect completed futures under mutex */
#define MAX_COMPLETED_BATCH 16
async_pending_t *completed[MAX_COMPLETED_BATCH];
int num_completed = 0;
pthread_mutex_lock(&worker->queue_mutex);
async_pending_t *p = worker->pending_head;
while (p != NULL && num_completed < MAX_COMPLETED_BATCH) {
if (p->future != NULL) {
/* Quick check if future is done (still needs GIL, but mutex held briefly) */
PyObject *done = PyObject_CallMethod(p->future, "done", NULL);
if (done != NULL && PyObject_IsTrue(done)) {
Py_DECREF(done);
completed[num_completed++] = p;
} else {
Py_XDECREF(done);
}
}
p = p->next;
}
pthread_mutex_unlock(&worker->queue_mutex);
/* Phase 2: Process completed callbacks outside mutex (no deadlock risk) */
for (int i = 0; i < num_completed; i++) {
async_future_callback(worker, completed[i]);
}
/* Phase 3: Remove processed items under mutex */
if (num_completed > 0) {
pthread_mutex_lock(&worker->queue_mutex);
for (int i = 0; i < num_completed; i++) {
async_pending_t *to_remove = completed[i];
/* Find and remove from list */
async_pending_t *prev = NULL;
p = worker->pending_head;
while (p != NULL) {
if (p == to_remove) {
/* Remove from list */
if (prev == NULL) {
worker->pending_head = p->next;
} else {
prev->next = p->next;
}
if (p == worker->pending_tail) {
worker->pending_tail = prev;
}
break;
}
prev = p;
p = p->next;
}
/* Clean up */
Py_DECREF(to_remove->future);
enif_free(to_remove);
}
pthread_mutex_unlock(&worker->queue_mutex);
}
}
/* Stop and close the event loop */
PyObject_CallMethod(loop, "stop", NULL);
PyObject_CallMethod(loop, "close", NULL);
Py_DECREF(loop);
worker->loop_running = false;
PyGILState_Release(gstate);
return NULL;
}
/* ============================================================================
* Resume callback NIFs
* ============================================================================ */
/* Forward declaration for the dirty resume NIF */
static ERL_NIF_TERM nif_resume_callback_dirty(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[]);
/**
* Resume a suspended callback by storing the result and scheduling replay.
*
* Args: StateRef, ResultBinary
*
* This NIF stores the callback result in the suspended state and schedules
* a dirty NIF (nif_resume_callback_dirty) to replay the Python code.
*/
static ERL_NIF_TERM nif_resume_callback(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[]) {
(void)argc;
suspended_state_t *state;
ErlNifBinary result_bin;
if (!enif_get_resource(env, argv[0], SUSPENDED_STATE_RESOURCE_TYPE, (void **)&state)) {
return make_error(env, "invalid_state_ref");
}
if (!enif_inspect_binary(env, argv[1], &result_bin)) {
return make_error(env, "invalid_result");
}
/* Store the result in the suspended state */
pthread_mutex_lock(&state->mutex);
/* Copy result data */
state->result_data = enif_alloc(result_bin.size);
if (state->result_data == NULL) {
pthread_mutex_unlock(&state->mutex);
return make_error(env, "alloc_failed");
}
memcpy(state->result_data, result_bin.data, result_bin.size);
state->result_len = result_bin.size;
state->has_result = true;
state->is_error = false;
pthread_mutex_unlock(&state->mutex);
/*
* Schedule the dirty resume NIF.
* This allows the current NIF to return immediately, and the dirty NIF
* will handle the Python replay on a dirty scheduler.
*/
ERL_NIF_TERM new_argv[1] = { argv[0] }; /* Pass StateRef to dirty NIF */
return enif_schedule_nif(env, "resume_callback_dirty",
ERL_NIF_DIRTY_JOB_IO_BOUND, nif_resume_callback_dirty, 1, new_argv);
}
/**
* Dirty NIF that replays Python code with the cached callback result.
*
* This is scheduled by nif_resume_callback and runs on a dirty I/O scheduler.
* It sets tl_current_suspended so erlang_call_impl can return the cached result,
* then re-runs the original Python code. When Python hits erlang.call() again,
* it gets the cached result and continues normally.
*/
static ERL_NIF_TERM nif_resume_callback_dirty(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[]) {
(void)argc;
suspended_state_t *state;
if (!enif_get_resource(env, argv[0], SUSPENDED_STATE_RESOURCE_TYPE, (void **)&state)) {
return make_error(env, "invalid_state_ref");
}
/* Verify the state has a result */
if (!state->has_result) {
return make_error(env, "no_result");
}
/* Set up thread-local state for replay */
tl_current_worker = state->worker;
tl_callback_env = env;
tl_current_suspended = state; /* erlang_call_impl will check this */
tl_allow_suspension = true;
ERL_NIF_TERM result;
if (state->request_type == PY_REQ_CALL) {
/* Replay a py:call */
char *module_name = enif_alloc(state->orig_module.size + 1);
char *func_name = enif_alloc(state->orig_func.size + 1);
if (module_name == NULL || func_name == NULL) {
enif_free(module_name);
enif_free(func_name);
tl_current_suspended = NULL;
return make_error(env, "alloc_failed");
}
memcpy(module_name, state->orig_module.data, state->orig_module.size);
module_name[state->orig_module.size] = '\0';
memcpy(func_name, state->orig_func.data, state->orig_func.size);
func_name[state->orig_func.size] = '\0';
PyGILState_STATE gstate = PyGILState_Ensure();
PyObject *func = NULL;
/* Get the function (same logic as process_request) */
if (strcmp(module_name, "__main__") == 0) {
func = PyDict_GetItemString(state->worker->locals, func_name);
if (func == NULL) {
func = PyDict_GetItemString(state->worker->globals, func_name);
}
if (func != NULL) {
Py_INCREF(func);
} else {
PyErr_Format(PyExc_NameError, "name '%s' is not defined", func_name);
result = make_py_error(env);
goto call_cleanup;
}
} else {
PyObject *module = PyImport_ImportModule(module_name);
if (module == NULL) {
result = make_py_error(env);
goto call_cleanup;
}
func = PyObject_GetAttrString(module, func_name);
Py_DECREF(module);
}
if (func == NULL) {
result = make_py_error(env);
goto call_cleanup;
}
/* Convert args */
unsigned int args_len;
if (!enif_get_list_length(state->orig_env, state->orig_args, &args_len)) {
Py_DECREF(func);
result = make_error(env, "invalid_args");
goto call_cleanup;
}
PyObject *args = PyTuple_New(args_len);
ERL_NIF_TERM head, tail = state->orig_args;
for (unsigned int i = 0; i < args_len; i++) {
enif_get_list_cell(state->orig_env, tail, &head, &tail);
PyObject *arg = term_to_py(state->orig_env, head);
if (arg == NULL) {
Py_DECREF(args);
Py_DECREF(func);
result = make_error(env, "arg_conversion_failed");
goto call_cleanup;
}
PyTuple_SET_ITEM(args, i, arg);
}
/* Convert kwargs */
PyObject *kwargs = NULL;
if (enif_is_map(state->orig_env, state->orig_kwargs)) {
kwargs = term_to_py(state->orig_env, state->orig_kwargs);
}
/* Call the function (this will hit erlang.call which returns cached result) */
PyObject *py_result = PyObject_Call(func, args, kwargs);
Py_DECREF(func);
Py_DECREF(args);
Py_XDECREF(kwargs);
if (py_result == NULL) {
if (tl_pending_callback) {
/*
* Flag-based callback detection during replay.
* Check flag FIRST, not exception type - this works even if
* Python code caught and re-raised the exception.
*/
PyErr_Clear(); /* Clear whatever exception is set */
/* Build exc_args tuple from thread-local storage */
PyObject *exc_args = build_pending_callback_exc_args();
if (exc_args == NULL) {
result = make_error(env, "build_exc_args_failed");
} else {
suspended_state_t *new_suspended = create_suspended_state_from_existing(env, exc_args, state);
Py_DECREF(exc_args);
if (new_suspended == NULL) {
tl_pending_callback = false;
result = make_error(env, "create_nested_suspended_state_failed");
} else {
result = build_suspended_result(env, new_suspended);
}
}
} else {
result = make_py_error(env);
}
} else {
ERL_NIF_TERM term_result = py_to_term(env, py_result);
Py_DECREF(py_result);
result = enif_make_tuple2(env, ATOM_OK, term_result);
}
call_cleanup:
PyGILState_Release(gstate);
enif_free(module_name);
enif_free(func_name);
} else if (state->request_type == PY_REQ_EVAL) {
/* Replay a py:eval */
char *code = enif_alloc(state->orig_code.size + 1);
if (code == NULL) {
tl_current_suspended = NULL;
return make_error(env, "alloc_failed");
}
memcpy(code, state->orig_code.data, state->orig_code.size);
code[state->orig_code.size] = '\0';
PyGILState_STATE gstate = PyGILState_Ensure();
/* Update locals if provided */
if (enif_is_map(state->orig_env, state->orig_locals)) {
PyObject *new_locals = term_to_py(state->orig_env, state->orig_locals);
if (new_locals != NULL && PyDict_Check(new_locals)) {
PyDict_Update(state->worker->locals, new_locals);
Py_DECREF(new_locals);
}
}
/* Compile and evaluate */
PyObject *compiled = Py_CompileString(code, "<erlang>", Py_eval_input);
if (compiled == NULL) {
result = make_py_error(env);
} else {
PyObject *py_result = PyEval_EvalCode(compiled, state->worker->globals,
state->worker->locals);
Py_DECREF(compiled);
if (py_result == NULL) {
if (tl_pending_callback) {
/*
* Flag-based callback detection during eval replay.
* Check flag FIRST, not exception type - this works even if
* Python code caught and re-raised the exception.
*/
PyErr_Clear(); /* Clear whatever exception is set */
/* Build exc_args tuple from thread-local storage */
PyObject *exc_args = build_pending_callback_exc_args();
if (exc_args == NULL) {
result = make_error(env, "build_exc_args_failed");
} else {
suspended_state_t *new_suspended = create_suspended_state_from_existing(env, exc_args, state);
Py_DECREF(exc_args);
if (new_suspended == NULL) {
tl_pending_callback = false;
result = make_error(env, "create_nested_suspended_state_failed");
} else {
result = build_suspended_result(env, new_suspended);
}
}
} else {
result = make_py_error(env);
}
} else {
ERL_NIF_TERM term_result = py_to_term(env, py_result);
Py_DECREF(py_result);
result = enif_make_tuple2(env, ATOM_OK, term_result);
}
}
PyGILState_Release(gstate);
enif_free(code);
} else {
result = make_error(env, "unsupported_request_type");
}
/* Clear thread-local state */
tl_current_worker = NULL;
tl_callback_env = NULL;
tl_current_suspended = NULL;
tl_allow_suspension = false;
return result;
}
/* ============================================================================
* NIF functions for callback name registration
* ============================================================================ */
/**
* @brief NIF to register a callback name in the C-side registry
*
* This allows the erlang module's __getattr__ to return ErlangFunction
* wrappers only for registered callbacks, preventing introspection issues.
*
* Args: Name (binary or atom)
* Returns: ok | {error, Reason}
*/
static ERL_NIF_TERM nif_register_callback_name(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[]) {
(void)argc;
ErlNifBinary name_bin;
char atom_buf[256];
const char *name;
size_t name_len;
if (enif_inspect_binary(env, argv[0], &name_bin)) {
name = (const char *)name_bin.data;
name_len = name_bin.size;
} else if (enif_get_atom(env, argv[0], atom_buf, sizeof(atom_buf), ERL_NIF_LATIN1)) {
name = atom_buf;
name_len = strlen(atom_buf);
} else {
return make_error(env, "invalid_name");
}
if (register_callback_name(name, name_len) < 0) {
return make_error(env, "registration_failed");
}
return ATOM_OK;
}
/**
* @brief NIF to unregister a callback name from the C-side registry
*
* Args: Name (binary or atom)
* Returns: ok
*/
static ERL_NIF_TERM nif_unregister_callback_name(ErlNifEnv *env, int argc, const ERL_NIF_TERM argv[]) {
(void)argc;
ErlNifBinary name_bin;
char atom_buf[256];
const char *name;
size_t name_len;
if (enif_inspect_binary(env, argv[0], &name_bin)) {
name = (const char *)name_bin.data;
name_len = name_bin.size;
} else if (enif_get_atom(env, argv[0], atom_buf, sizeof(atom_buf), ERL_NIF_LATIN1)) {
name = atom_buf;
name_len = strlen(atom_buf);
} else {
return make_error(env, "invalid_name");
}
unregister_callback_name(name, name_len);
return ATOM_OK;
}