/* * 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) */ /* ============================================================================ * Cached Python Function References * * Cache frequently-used Python functions to avoid repeated module import * and attribute lookup overhead on every callback. * ============================================================================ */ /** @brief Cached reference to ast.literal_eval function */ static PyObject *g_ast_literal_eval = NULL; /** * @brief Initialize cached Python function references * * Called during module initialization. Must be called with GIL held. */ static void init_callback_cache(void) { if (g_ast_literal_eval == NULL) { PyObject *ast_mod = PyImport_ImportModule("ast"); if (ast_mod != NULL) { g_ast_literal_eval = PyObject_GetAttrString(ast_mod, "literal_eval"); Py_DECREF(ast_mod); } if (g_ast_literal_eval == NULL) { PyErr_Clear(); /* Non-fatal if unavailable */ } } } /** * @brief Cleanup cached Python function references * * Called during module cleanup. Must be called with GIL held. */ static void cleanup_callback_cache(void) { Py_XDECREF(g_ast_literal_eval); g_ast_literal_eval = NULL; } /* ============================================================================ * 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 using cached function */ if (g_ast_literal_eval != NULL) { PyObject *arg = PyUnicode_FromStringAndSize(result_str, result_len); if (arg != NULL) { result = PyObject_CallFunctionObjArgs(g_ast_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); } } } 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("", 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 cached Python function references */ init_callback_cache(); /* 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, "", 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; }