/* * 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_subinterp_thread.c * @brief OWN_GIL subinterpreter thread pool implementation * @author Benoit Chesneau * * Implements a pthread pool where each thread owns a Python subinterpreter * with OWN_GIL for true parallelism. */ #include "py_subinterp_thread.h" #include "py_nif.h" #include "py_buffer.h" #include #include #include #include #ifdef HAVE_SUBINTERPRETERS /* ============================================================================ * Global State * ============================================================================ */ /** @brief Global thread pool instance */ subinterp_thread_pool_t g_thread_pool = {0}; /* Forward declarations */ static void *worker_thread_main(void *arg); static int worker_create_namespace(subinterp_thread_worker_t *w, uint64_t handle_id); static void worker_destroy_namespace(subinterp_thread_worker_t *w, uint64_t handle_id); static subinterp_namespace_t *worker_find_namespace(subinterp_thread_worker_t *w, uint64_t handle_id); static int write_full(int fd, const void *buf, size_t count); static int read_full(int fd, void *buf, size_t count); /* Defined in py_callback.c */ extern int create_erlang_module(void); /* ============================================================================ * Pool Management * ============================================================================ */ int subinterp_thread_pool_init(int num_workers) { if (atomic_load(&g_thread_pool.initialized)) { return 0; /* Already initialized */ } /* Set default/cap worker count */ if (num_workers <= 0) { num_workers = SUBINTERP_THREAD_POOL_DEFAULT; } if (num_workers > SUBINTERP_THREAD_POOL_MAX) { num_workers = SUBINTERP_THREAD_POOL_MAX; } /* Initialize pool state */ memset(&g_thread_pool, 0, sizeof(g_thread_pool)); g_thread_pool.num_workers = num_workers; atomic_store(&g_thread_pool.next_worker, 0); atomic_store(&g_thread_pool.next_handle_id, 1); atomic_store(&g_thread_pool.next_request_id, 1); /* Create workers - declare i outside loop for cleanup_workers label */ int i; for (i = 0; i < num_workers; i++) { subinterp_thread_worker_t *w = &g_thread_pool.workers[i]; w->worker_id = i; atomic_store(&w->running, false); atomic_store(&w->shutdown_requested, false); atomic_store(&w->initialized, false); atomic_store(&w->requests_processed, 0); atomic_store(&w->errors_count, 0); w->num_namespaces = 0; /* Initialize mutexes */ if (pthread_mutex_init(&w->dispatch_mutex, NULL) != 0) { fprintf(stderr, "subinterp_thread_pool_init: failed to init dispatch_mutex for worker %d\n", i); goto cleanup_workers; } if (pthread_mutex_init(&w->ns_mutex, NULL) != 0) { fprintf(stderr, "subinterp_thread_pool_init: failed to init ns_mutex for worker %d\n", i); pthread_mutex_destroy(&w->dispatch_mutex); goto cleanup_workers; } /* Create pipes */ if (pipe(w->cmd_pipe) < 0) { fprintf(stderr, "subinterp_thread_pool_init: failed to create cmd_pipe for worker %d: %s\n", i, strerror(errno)); pthread_mutex_destroy(&w->dispatch_mutex); pthread_mutex_destroy(&w->ns_mutex); goto cleanup_workers; } /* Non-blocking cmd_pipe write end so the dispatch NIFs can bound their * writes with write_all_with_deadline (no dirty-scheduler stall on a full * pipe). The worker thread reads cmd_pipe[0] blocking, which is fine. */ { int wfl = fcntl(w->cmd_pipe[1], F_GETFL, 0); if (wfl >= 0) (void)fcntl(w->cmd_pipe[1], F_SETFL, wfl | O_NONBLOCK); } if (pipe(w->result_pipe) < 0) { fprintf(stderr, "subinterp_thread_pool_init: failed to create result_pipe for worker %d: %s\n", i, strerror(errno)); close(w->cmd_pipe[0]); close(w->cmd_pipe[1]); pthread_mutex_destroy(&w->dispatch_mutex); pthread_mutex_destroy(&w->ns_mutex); goto cleanup_workers; } /* Set non-blocking on read ends for proper timeout handling */ /* Actually, keep blocking for simplicity - we control timeouts via protocol */ /* Start worker thread */ if (pthread_create(&w->thread, NULL, worker_thread_main, w) != 0) { fprintf(stderr, "subinterp_thread_pool_init: failed to create thread for worker %d: %s\n", i, strerror(errno)); close(w->cmd_pipe[0]); close(w->cmd_pipe[1]); close(w->result_pipe[0]); close(w->result_pipe[1]); pthread_mutex_destroy(&w->dispatch_mutex); pthread_mutex_destroy(&w->ns_mutex); goto cleanup_workers; } /* Wait for worker to initialize */ int timeout_ms = 5000; /* 5 second timeout */ int waited = 0; while (!atomic_load(&w->initialized) && waited < timeout_ms) { usleep(10000); /* 10ms */ waited += 10; } if (!atomic_load(&w->initialized)) { fprintf(stderr, "subinterp_thread_pool_init: worker %d failed to initialize\n", i); atomic_store(&w->shutdown_requested, true); /* Write shutdown message to unblock worker if stuck */ owngil_header_t shutdown = { .magic = OWNGIL_MAGIC, .version = OWNGIL_PROTOCOL_VERSION, .msg_type = MSG_REQUEST, .req_type = REQ_SHUTDOWN, .payload_len = 0, }; write(w->cmd_pipe[1], &shutdown, sizeof(shutdown)); pthread_join(w->thread, NULL); close(w->cmd_pipe[0]); close(w->cmd_pipe[1]); close(w->result_pipe[0]); close(w->result_pipe[1]); pthread_mutex_destroy(&w->dispatch_mutex); pthread_mutex_destroy(&w->ns_mutex); goto cleanup_workers; } } atomic_store(&g_thread_pool.initialized, true); #ifdef DEBUG fprintf(stderr, "subinterp_thread_pool_init: created %d OWN_GIL workers\n", num_workers); #endif return 0; cleanup_workers: /* Clean up already created workers */ for (int j = 0; j < i; j++) { subinterp_thread_worker_t *w = &g_thread_pool.workers[j]; atomic_store(&w->shutdown_requested, true); owngil_header_t shutdown = { .magic = OWNGIL_MAGIC, .version = OWNGIL_PROTOCOL_VERSION, .msg_type = MSG_REQUEST, .req_type = REQ_SHUTDOWN, .payload_len = 0, }; write(w->cmd_pipe[1], &shutdown, sizeof(shutdown)); pthread_join(w->thread, NULL); close(w->cmd_pipe[0]); close(w->cmd_pipe[1]); close(w->result_pipe[0]); close(w->result_pipe[1]); pthread_mutex_destroy(&w->dispatch_mutex); pthread_mutex_destroy(&w->ns_mutex); } return -1; } void subinterp_thread_pool_shutdown(void) { if (!atomic_load(&g_thread_pool.initialized)) { return; } /* Mark as not initialized to prevent new work */ atomic_store(&g_thread_pool.initialized, false); /* Signal all workers to shut down */ for (int i = 0; i < g_thread_pool.num_workers; i++) { subinterp_thread_worker_t *w = &g_thread_pool.workers[i]; if (!atomic_load(&w->running)) { continue; } atomic_store(&w->shutdown_requested, true); /* Send shutdown message */ owngil_header_t shutdown = { .magic = OWNGIL_MAGIC, .version = OWNGIL_PROTOCOL_VERSION, .msg_type = MSG_REQUEST, .req_type = REQ_SHUTDOWN, .payload_len = 0, }; write_full(w->cmd_pipe[1], &shutdown, sizeof(shutdown)); } /* Wait for all workers to exit */ for (int i = 0; i < g_thread_pool.num_workers; i++) { subinterp_thread_worker_t *w = &g_thread_pool.workers[i]; if (w->thread != 0) { pthread_join(w->thread, NULL); } /* Close pipes */ if (w->cmd_pipe[0] >= 0) close(w->cmd_pipe[0]); if (w->cmd_pipe[1] >= 0) close(w->cmd_pipe[1]); if (w->result_pipe[0] >= 0) close(w->result_pipe[0]); if (w->result_pipe[1] >= 0) close(w->result_pipe[1]); /* Destroy mutexes */ pthread_mutex_destroy(&w->dispatch_mutex); pthread_mutex_destroy(&w->ns_mutex); } g_thread_pool.num_workers = 0; #ifdef DEBUG fprintf(stderr, "subinterp_thread_pool_shutdown: complete\n"); #endif } bool subinterp_thread_pool_is_ready(void) { return atomic_load(&g_thread_pool.initialized); } void subinterp_thread_pool_stats(int *num_workers, uint64_t *total_requests, uint64_t *total_errors) { if (num_workers) *num_workers = g_thread_pool.num_workers; uint64_t reqs = 0, errs = 0; for (int i = 0; i < g_thread_pool.num_workers; i++) { reqs += atomic_load(&g_thread_pool.workers[i].requests_processed); errs += atomic_load(&g_thread_pool.workers[i].errors_count); } if (total_requests) *total_requests = reqs; if (total_errors) *total_errors = errs; } /* ============================================================================ * Worker Thread Main Loop * ============================================================================ */ static void *worker_thread_main(void *arg) { subinterp_thread_worker_t *w = (subinterp_thread_worker_t *)arg; /* Create OWN_GIL subinterpreter. * For OWN_GIL, we need the main GIL to create the subinterpreter, * then the subinterpreter gets its own GIL. After creation, * we're switched to the new subinterpreter's thread state. */ PyInterpreterConfig config = { .use_main_obmalloc = 0, .allow_fork = 0, .allow_exec = 0, .allow_threads = 1, .allow_daemon_threads = 0, .check_multi_interp_extensions = 1, .gil = PyInterpreterConfig_OWN_GIL, }; /* Acquire main GIL to create subinterpreter */ PyGILState_STATE gstate = PyGILState_Ensure(); /* Save main thread state before creating subinterpreter */ PyThreadState *main_tstate = PyThreadState_Get(); PyStatus status = Py_NewInterpreterFromConfig(&w->tstate, &config); if (PyStatus_Exception(status) || w->tstate == NULL) { fprintf(stderr, "worker %d: failed to create OWN_GIL subinterpreter\n", w->worker_id); /* Restore main thread state and release */ PyThreadState_Swap(main_tstate); PyGILState_Release(gstate); return NULL; } /* Now we're in the new subinterpreter's thread state with its own GIL. * The main GIL was released when Py_NewInterpreterFromConfig switched to OWN_GIL. */ w->interp = PyThreadState_GetInterpreter(w->tstate); /* Create erlang module in this subinterpreter */ if (create_erlang_module() < 0) { fprintf(stderr, "worker %d: failed to create erlang module\n", w->worker_id); PyErr_Clear(); /* Continue without erlang module - callbacks won't work */ } else { /* Register PyBuffer with erlang module in this subinterpreter */ if (PyBuffer_register_with_module() < 0) { PyErr_Clear(); /* Non-fatal - PyBuffer just won't be available */ } } /* Initialize asyncio for this worker */ w->asyncio_module = PyImport_ImportModule("asyncio"); if (w->asyncio_module == NULL) { fprintf(stderr, "worker %d: failed to import asyncio\n", w->worker_id); PyErr_Clear(); } else { /* Create a new event loop for this worker */ PyObject *new_event_loop = PyObject_CallMethod(w->asyncio_module, "new_event_loop", NULL); if (new_event_loop == NULL) { fprintf(stderr, "worker %d: failed to create asyncio event loop\n", w->worker_id); PyErr_Clear(); } else { w->asyncio_loop = new_event_loop; /* Set as the running event loop for this thread */ PyObject *result = PyObject_CallMethod(w->asyncio_module, "set_event_loop", "O", w->asyncio_loop); Py_XDECREF(result); PyErr_Clear(); } } /* Release the subinterpreter's GIL (we'll acquire it per-request) */ PyEval_SaveThread(); /* Signal that we're initialized */ atomic_store(&w->running, true); atomic_store(&w->initialized, true); /* Main command loop */ while (!atomic_load(&w->shutdown_requested)) { owngil_header_t header; int n = read_full(w->cmd_pipe[0], &header, sizeof(header)); if (n <= 0) { if (errno == EINTR) continue; break; /* Pipe closed or error */ } /* Validate header */ if (header.magic != OWNGIL_MAGIC || header.version != OWNGIL_PROTOCOL_VERSION) { fprintf(stderr, "worker %d: invalid protocol header\n", w->worker_id); continue; } /* Handle shutdown */ if (header.req_type == REQ_SHUTDOWN) { break; } /* Read payload if present */ unsigned char *payload = NULL; if (header.payload_len > 0) { payload = malloc(header.payload_len); if (payload == NULL) { fprintf(stderr, "worker %d: failed to allocate payload\n", w->worker_id); continue; } n = read_full(w->cmd_pipe[0], payload, header.payload_len); if (n != (int)header.payload_len) { fprintf(stderr, "worker %d: failed to read payload\n", w->worker_id); free(payload); continue; } } /* Acquire our GIL for Python execution */ PyEval_RestoreThread(w->tstate); /* Handle namespace management (needs GIL for Python dict operations) */ if (header.req_type == REQ_CREATE_NS) { worker_create_namespace(w, header.handle_id); PyEval_SaveThread(); /* Send simple OK response */ owngil_header_t resp = { .magic = OWNGIL_MAGIC, .version = OWNGIL_PROTOCOL_VERSION, .msg_type = MSG_RESPONSE, .request_id = header.request_id, .payload_len = 0, }; write_full(w->result_pipe[1], &resp, sizeof(resp)); free(payload); continue; } if (header.req_type == REQ_DESTROY_NS) { worker_destroy_namespace(w, header.handle_id); PyEval_SaveThread(); /* Send simple OK response */ owngil_header_t resp = { .magic = OWNGIL_MAGIC, .version = OWNGIL_PROTOCOL_VERSION, .msg_type = MSG_RESPONSE, .request_id = header.request_id, .payload_len = 0, }; write_full(w->result_pipe[1], &resp, sizeof(resp)); free(payload); continue; } /* Handle apply imports - imports modules into sys.modules */ if (header.req_type == REQ_APPLY_IMPORTS) { /* Payload is ETF list of {ModuleBin, FuncBin | all} tuples */ if (payload != NULL && header.payload_len > 0) { ErlNifEnv *tmp_env = enif_alloc_env(); if (tmp_env != NULL) { ERL_NIF_TERM imports_list; if (enif_binary_to_term(tmp_env, payload, header.payload_len, &imports_list, ERL_NIF_BIN2TERM_SAFE) != 0) { ERL_NIF_TERM head, tail = imports_list; int arity; const ERL_NIF_TERM *tuple; while (enif_get_list_cell(tmp_env, tail, &head, &tail)) { if (enif_get_tuple(tmp_env, head, &arity, &tuple) && arity == 2) { ErlNifBinary module_bin; if (enif_inspect_binary(tmp_env, tuple[0], &module_bin)) { char *module_name = enif_alloc(module_bin.size + 1); if (module_name != NULL) { memcpy(module_name, module_bin.data, module_bin.size); module_name[module_bin.size] = '\0'; /* Skip __main__ */ if (strcmp(module_name, "__main__") != 0) { PyObject *mod = PyImport_ImportModule(module_name); if (mod != NULL) { Py_DECREF(mod); } else { PyErr_Clear(); } } enif_free(module_name); } } } } } enif_free_env(tmp_env); } } PyEval_SaveThread(); /* Send OK response */ owngil_header_t resp = { .magic = OWNGIL_MAGIC, .version = OWNGIL_PROTOCOL_VERSION, .msg_type = MSG_RESPONSE, .request_id = header.request_id, .payload_len = 0, }; write_full(w->result_pipe[1], &resp, sizeof(resp)); free(payload); continue; } /* Handle apply paths - add paths to sys.path */ if (header.req_type == REQ_APPLY_PATHS) { /* Payload is ETF list of path binaries */ if (payload != NULL && header.payload_len > 0) { ErlNifEnv *tmp_env = enif_alloc_env(); if (tmp_env != NULL) { ERL_NIF_TERM paths_list; if (enif_binary_to_term(tmp_env, payload, header.payload_len, &paths_list, ERL_NIF_BIN2TERM_SAFE) != 0) { PyObject *sys_path = PySys_GetObject("path"); if (sys_path != NULL && PyList_Check(sys_path)) { ERL_NIF_TERM head, tail = paths_list; /* Insert in reverse order so first path ends up first */ while (enif_get_list_cell(tmp_env, tail, &head, &tail)) { ErlNifBinary path_bin; if (enif_inspect_binary(tmp_env, head, &path_bin)) { PyObject *path_str = PyUnicode_FromStringAndSize( (const char *)path_bin.data, path_bin.size); if (path_str != NULL) { /* Check if path already in sys.path */ int contains = PySequence_Contains(sys_path, path_str); if (contains == 0) { PyList_Insert(sys_path, 0, path_str); } Py_DECREF(path_str); } } } } } enif_free_env(tmp_env); } } PyEval_SaveThread(); /* Send OK response */ owngil_header_t resp = { .magic = OWNGIL_MAGIC, .version = OWNGIL_PROTOCOL_VERSION, .msg_type = MSG_RESPONSE, .request_id = header.request_id, .payload_len = 0, }; write_full(w->result_pipe[1], &resp, sizeof(resp)); free(payload); continue; } /* Find namespace for this handle */ subinterp_namespace_t *ns = worker_find_namespace(w, header.handle_id); if (ns == NULL) { /* Namespace not found - create default one on the fly */ worker_create_namespace(w, header.handle_id); ns = worker_find_namespace(w, header.handle_id); } /* Process the request */ owngil_header_t resp_header = { .magic = OWNGIL_MAGIC, .version = OWNGIL_PROTOCOL_VERSION, .msg_type = MSG_RESPONSE, .request_id = header.request_id, .payload_len = 0, }; unsigned char *resp_payload = NULL; size_t resp_payload_len = 0; /* Decode payload using temporary env */ ErlNifEnv *tmp_env = enif_alloc_env(); ERL_NIF_TERM payload_term; int arity; const ERL_NIF_TERM *elements; bool success = false; if (tmp_env != NULL && header.payload_len > 0) { if (enif_binary_to_term(tmp_env, payload, header.payload_len, &payload_term, ERL_NIF_BIN2TERM_SAFE) != 0) { if (enif_get_tuple(tmp_env, payload_term, &arity, &elements)) { /* Execute based on request type */ PyObject *result = NULL; PyObject *globals = ns ? ns->globals : PyDict_New(); PyObject *locals = ns ? ns->locals : PyDict_New(); bool owns_globals = (ns == NULL); bool owns_locals = (ns == NULL); /* Check allocation if we own the dicts */ if ((owns_globals && globals == NULL) || (owns_locals && locals == NULL)) { if (owns_globals) Py_XDECREF(globals); if (owns_locals) Py_XDECREF(locals); /* Per-request dict allocation failure: respond with an * error and keep serving. This previously `break`ed the * worker command loop, permanently killing the thread (and * leaving the GIL held), wedging every session routed to it. */ resp_header.msg_type = MSG_ERROR; goto send_response; } switch (header.req_type) { case REQ_CALL: case REQ_CAST: { /* Payload: {Module, Func, Args, Kwargs} */ if (arity >= 3) { ErlNifBinary mod_bin, func_bin; char mod_str[256], func_str[256]; /* Get module name */ if (enif_inspect_binary(tmp_env, elements[0], &mod_bin)) { size_t len = mod_bin.size < 255 ? mod_bin.size : 255; memcpy(mod_str, mod_bin.data, len); mod_str[len] = '\0'; } else if (enif_get_atom(tmp_env, elements[0], mod_str, 256, ERL_NIF_LATIN1)) { /* Already filled */ } else { if (owns_globals) Py_DECREF(globals); if (owns_locals) Py_DECREF(locals); break; } /* Get function name */ if (enif_inspect_binary(tmp_env, elements[1], &func_bin)) { size_t len = func_bin.size < 255 ? func_bin.size : 255; memcpy(func_str, func_bin.data, len); func_str[len] = '\0'; } else if (enif_get_atom(tmp_env, elements[1], func_str, 256, ERL_NIF_LATIN1)) { /* Already filled */ } else { if (owns_globals) Py_DECREF(globals); if (owns_locals) Py_DECREF(locals); break; } /* Import module */ PyObject *module = NULL; if (ns && ns->module_cache) { PyObject *key = PyUnicode_FromString(mod_str); module = PyDict_GetItem(ns->module_cache, key); if (module == NULL) { module = PyImport_ImportModule(mod_str); if (module) { PyDict_SetItem(ns->module_cache, key, module); } } else { Py_INCREF(module); } Py_DECREF(key); } else { module = PyImport_ImportModule(mod_str); } if (module == NULL) { PyErr_Clear(); if (owns_globals) Py_DECREF(globals); if (owns_locals) Py_DECREF(locals); break; } /* Get function */ PyObject *func = PyObject_GetAttrString(module, func_str); Py_DECREF(module); if (func == NULL) { PyErr_Clear(); if (owns_globals) Py_DECREF(globals); if (owns_locals) Py_DECREF(locals); break; } /* Convert args list to Python tuple */ ERL_NIF_TERM args_list = elements[2]; unsigned int args_len; PyObject *py_args = NULL; if (enif_get_list_length(tmp_env, args_list, &args_len)) { py_args = PyTuple_New(args_len); if (py_args) { ERL_NIF_TERM head, tail = args_list; for (unsigned int idx = 0; idx < args_len; idx++) { if (!enif_get_list_cell(tmp_env, tail, &head, &tail)) { Py_DECREF(py_args); py_args = NULL; break; } PyObject *py_arg = term_to_py(tmp_env, head); if (py_arg == NULL) { Py_DECREF(py_args); py_args = NULL; break; } PyTuple_SET_ITEM(py_args, idx, py_arg); } } } if (py_args == NULL) { py_args = PyTuple_New(0); } /* Call function */ result = PyObject_Call(func, py_args, NULL); Py_DECREF(py_args); Py_DECREF(func); if (result == NULL) { PyErr_Clear(); } else { success = true; } } break; } case REQ_ASYNC_CALL: { /* Payload: {Module, Func, Args, Kwargs, CallerPid, Ref} */ /* For async calls, we run the coroutine and send result via erlang.send() */ if (arity >= 6) { ErlNifBinary mod_bin, func_bin; char mod_str[256], func_str[256]; /* Get module name */ if (enif_inspect_binary(tmp_env, elements[0], &mod_bin)) { size_t len = mod_bin.size < 255 ? mod_bin.size : 255; memcpy(mod_str, mod_bin.data, len); mod_str[len] = '\0'; } else if (enif_get_atom(tmp_env, elements[0], mod_str, 256, ERL_NIF_LATIN1)) { /* Already filled */ } else { if (owns_globals) Py_DECREF(globals); if (owns_locals) Py_DECREF(locals); break; } /* Get function name */ if (enif_inspect_binary(tmp_env, elements[1], &func_bin)) { size_t len = func_bin.size < 255 ? func_bin.size : 255; memcpy(func_str, func_bin.data, len); func_str[len] = '\0'; } else if (enif_get_atom(tmp_env, elements[1], func_str, 256, ERL_NIF_LATIN1)) { /* Already filled */ } else { if (owns_globals) Py_DECREF(globals); if (owns_locals) Py_DECREF(locals); break; } /* Import module */ PyObject *module = NULL; if (ns && ns->module_cache) { PyObject *key = PyUnicode_FromString(mod_str); module = PyDict_GetItem(ns->module_cache, key); if (module == NULL) { module = PyImport_ImportModule(mod_str); if (module) { PyDict_SetItem(ns->module_cache, key, module); } } else { Py_INCREF(module); } Py_DECREF(key); } else { module = PyImport_ImportModule(mod_str); } if (module == NULL) { PyErr_Clear(); if (owns_globals) Py_DECREF(globals); if (owns_locals) Py_DECREF(locals); break; } /* Get function */ PyObject *func = PyObject_GetAttrString(module, func_str); Py_DECREF(module); if (func == NULL) { PyErr_Clear(); if (owns_globals) Py_DECREF(globals); if (owns_locals) Py_DECREF(locals); break; } /* Convert args list to Python tuple */ ERL_NIF_TERM args_list = elements[2]; unsigned int args_len; PyObject *py_args = NULL; if (enif_get_list_length(tmp_env, args_list, &args_len)) { py_args = PyTuple_New(args_len); if (py_args) { ERL_NIF_TERM head, tail = args_list; for (unsigned int idx = 0; idx < args_len; idx++) { if (!enif_get_list_cell(tmp_env, tail, &head, &tail)) { Py_DECREF(py_args); py_args = NULL; break; } PyObject *py_arg = term_to_py(tmp_env, head); if (py_arg == NULL) { Py_DECREF(py_args); py_args = NULL; break; } PyTuple_SET_ITEM(py_args, idx, py_arg); } } } if (py_args == NULL) { py_args = PyTuple_New(0); } /* Call function */ result = PyObject_Call(func, py_args, NULL); Py_DECREF(py_args); Py_DECREF(func); if (result == NULL) { PyErr_Clear(); } else { /* Check if result is a coroutine and run it */ if (w->asyncio_loop != NULL && PyCoro_CheckExact(result)) { PyObject *final_result = PyObject_CallMethod( w->asyncio_loop, "run_until_complete", "O", result); Py_DECREF(result); result = final_result; if (result == NULL) { PyErr_Clear(); } } if (result != NULL) { success = true; } } /* Send result via erlang.send() to CallerPid */ /* elements[4] = CallerPid, elements[5] = Ref */ ERL_NIF_TERM result_term; if (success && result != NULL) { ERL_NIF_TERM py_result = py_to_term(tmp_env, result); result_term = enif_make_tuple2(tmp_env, enif_make_atom(tmp_env, "ok"), py_result); } else { result_term = enif_make_tuple2(tmp_env, enif_make_atom(tmp_env, "error"), enif_make_atom(tmp_env, "execution_failed")); } /* Build {async_result, Ref, Result} message */ ERL_NIF_TERM msg = enif_make_tuple3(tmp_env, enif_make_atom(tmp_env, "async_result"), elements[5], /* Ref */ result_term); /* Get CallerPid and send */ ErlNifPid caller_pid; if (enif_get_local_pid(tmp_env, elements[4], &caller_pid)) { enif_send(NULL, &caller_pid, tmp_env, msg); } Py_XDECREF(result); result = NULL; /* Don't process result in normal path */ success = false; /* Already handled */ } if (owns_globals) Py_DECREF(globals); if (owns_locals) Py_DECREF(locals); break; } case REQ_EVAL: { /* Payload: {Code, Locals} */ if (arity >= 1) { ErlNifBinary code_bin; if (enif_inspect_binary(tmp_env, elements[0], &code_bin)) { char *code_str = malloc(code_bin.size + 1); if (code_str) { memcpy(code_str, code_bin.data, code_bin.size); code_str[code_bin.size] = '\0'; result = PyRun_String(code_str, Py_eval_input, globals, locals); free(code_str); if (result == NULL) { PyErr_Clear(); } else { success = true; } } } } break; } case REQ_EXEC: { /* Payload: {Code} */ if (arity >= 1) { ErlNifBinary code_bin; if (enif_inspect_binary(tmp_env, elements[0], &code_bin)) { char *code_str = malloc(code_bin.size + 1); if (code_str) { memcpy(code_str, code_bin.data, code_bin.size); code_str[code_bin.size] = '\0'; result = PyRun_String(code_str, Py_file_input, globals, locals); free(code_str); if (result == NULL) { PyErr_Clear(); } else { Py_DECREF(result); result = Py_None; Py_INCREF(result); success = true; } } } } break; } default: break; } /* Clean up owned dicts after switch completes */ if (owns_globals) Py_DECREF(globals); if (owns_locals) Py_DECREF(locals); /* Serialize result using py_to_term for full type support */ if (success && result != NULL) { ERL_NIF_TERM result_term = py_to_term(tmp_env, result); Py_XDECREF(result); /* Wrap in {ok, Result} */ ERL_NIF_TERM ok_tuple = enif_make_tuple2(tmp_env, enif_make_atom(tmp_env, "ok"), result_term); /* Serialize to ETF */ ErlNifBinary etf_bin; if (enif_term_to_binary(tmp_env, ok_tuple, &etf_bin)) { resp_payload = malloc(etf_bin.size); if (resp_payload) { memcpy(resp_payload, etf_bin.data, etf_bin.size); resp_payload_len = etf_bin.size; } enif_release_binary(&etf_bin); } } else { resp_header.msg_type = MSG_ERROR; /* Serialize error */ ERL_NIF_TERM err_tuple = enif_make_tuple2(tmp_env, enif_make_atom(tmp_env, "error"), enif_make_atom(tmp_env, "execution_failed")); ErlNifBinary etf_bin; if (enif_term_to_binary(tmp_env, err_tuple, &etf_bin)) { resp_payload = malloc(etf_bin.size); if (resp_payload) { memcpy(resp_payload, etf_bin.data, etf_bin.size); resp_payload_len = etf_bin.size; } enif_release_binary(&etf_bin); } } } } } send_response: if (tmp_env) { enif_free_env(tmp_env); } /* Release GIL */ PyEval_SaveThread(); /* Send response (except for cast) */ if (header.req_type != REQ_CAST) { resp_header.payload_len = resp_payload_len; write_full(w->result_pipe[1], &resp_header, sizeof(resp_header)); if (resp_payload_len > 0) { write_full(w->result_pipe[1], resp_payload, resp_payload_len); } } free(payload); free(resp_payload); atomic_fetch_add(&w->requests_processed, 1); } /* Cleanup */ PyEval_RestoreThread(w->tstate); /* Clean up all namespaces */ pthread_mutex_lock(&w->ns_mutex); for (int i = 0; i < w->num_namespaces; i++) { subinterp_namespace_t *ns = &w->namespaces[i]; if (ns->initialized) { Py_XDECREF(ns->asyncio_loop); Py_XDECREF(ns->module_cache); Py_XDECREF(ns->globals); Py_XDECREF(ns->locals); } } w->num_namespaces = 0; pthread_mutex_unlock(&w->ns_mutex); /* Clean up worker asyncio resources */ Py_XDECREF(w->asyncio_loop); w->asyncio_loop = NULL; Py_XDECREF(w->asyncio_module); w->asyncio_module = NULL; /* End interpreter */ Py_EndInterpreter(w->tstate); w->tstate = NULL; w->interp = NULL; atomic_store(&w->running, false); return NULL; } /* ============================================================================ * Namespace Management * ============================================================================ */ static int worker_create_namespace(subinterp_thread_worker_t *w, uint64_t handle_id) { pthread_mutex_lock(&w->ns_mutex); /* Check if namespace already exists */ for (int i = 0; i < w->num_namespaces; i++) { if (w->namespaces[i].handle_id == handle_id && w->namespaces[i].initialized) { pthread_mutex_unlock(&w->ns_mutex); return 0; /* Already exists */ } } /* Find empty slot or add new */ int slot = -1; for (int i = 0; i < w->num_namespaces; i++) { if (!w->namespaces[i].initialized) { slot = i; break; } } if (slot < 0) { if (w->num_namespaces >= MAX_NAMESPACES_PER_WORKER) { pthread_mutex_unlock(&w->ns_mutex); return -1; /* Full */ } slot = w->num_namespaces++; } subinterp_namespace_t *ns = &w->namespaces[slot]; ns->handle_id = handle_id; ns->globals = PyDict_New(); ns->locals = PyDict_New(); ns->module_cache = PyDict_New(); ns->asyncio_loop = NULL; /* Uses worker's shared event loop */ memset(&ns->owner_pid, 0, sizeof(ns->owner_pid)); if (ns->globals && ns->locals && ns->module_cache) { /* Import __builtins__ */ PyObject *builtins = PyEval_GetBuiltins(); PyDict_SetItemString(ns->globals, "__builtins__", builtins); ns->initialized = true; } else { Py_XDECREF(ns->globals); Py_XDECREF(ns->locals); Py_XDECREF(ns->module_cache); ns->globals = NULL; ns->locals = NULL; ns->module_cache = NULL; pthread_mutex_unlock(&w->ns_mutex); return -1; } pthread_mutex_unlock(&w->ns_mutex); return 0; } static void worker_destroy_namespace(subinterp_thread_worker_t *w, uint64_t handle_id) { /* Note: Caller must hold the GIL */ pthread_mutex_lock(&w->ns_mutex); for (int i = 0; i < w->num_namespaces; i++) { if (w->namespaces[i].handle_id == handle_id && w->namespaces[i].initialized) { subinterp_namespace_t *ns = &w->namespaces[i]; Py_XDECREF(ns->module_cache); Py_XDECREF(ns->globals); Py_XDECREF(ns->locals); ns->initialized = false; ns->globals = NULL; ns->locals = NULL; ns->module_cache = NULL; break; } } pthread_mutex_unlock(&w->ns_mutex); } static subinterp_namespace_t *worker_find_namespace(subinterp_thread_worker_t *w, uint64_t handle_id) { pthread_mutex_lock(&w->ns_mutex); for (int i = 0; i < w->num_namespaces; i++) { if (w->namespaces[i].handle_id == handle_id && w->namespaces[i].initialized) { pthread_mutex_unlock(&w->ns_mutex); return &w->namespaces[i]; } } pthread_mutex_unlock(&w->ns_mutex); return NULL; } /* ============================================================================ * Utility Functions * ============================================================================ */ static int write_full(int fd, const void *buf, size_t count) { const unsigned char *p = buf; size_t remaining = count; while (remaining > 0) { ssize_t n = write(fd, p, remaining); if (n < 0) { if (errno == EINTR) continue; return -1; } p += n; remaining -= n; } return count; } static int read_full(int fd, void *buf, size_t count) { unsigned char *p = buf; size_t remaining = count; while (remaining > 0) { ssize_t n = read(fd, p, remaining); if (n < 0) { if (errno == EINTR) continue; return -1; } if (n == 0) { return count - remaining; /* EOF */ } p += n; remaining -= n; } return count; } #endif /* HAVE_SUBINTERPRETERS */