Add ThreadPoolExecutor support for erlang.call()

Python threads spawned via concurrent.futures.ThreadPoolExecutor can now
call erlang.call() without blocking. Each spawned thread lazily acquires
a dedicated "thread worker" channel with its own Erlang handler process.

Key components:
- c_src/py_thread_worker.c: Thread worker pool with pipe-based IPC
- src/py_thread_handler.erl: Coordinator and per-thread handlers
- Modified erlang_call_impl() to detect spawned threads
- Automatic cleanup via pthread_key_t destructor on thread exit

Test suite: test/py_thread_callback_SUITE.erl (6 tests)

Author: benoitc

CHANGELOG.md

@@ -4,6 +4,15 @@
 
 ### Added
 
+- **ThreadPoolExecutor Support** - Python threads spawned via `concurrent.futures.ThreadPoolExecutor`
+  can now call `erlang.call()` without blocking
+  - Each spawned thread lazily acquires a dedicated "thread worker" channel
+  - One lightweight Erlang process per Python thread handles callbacks
+  - Automatic cleanup when Python thread exits via `pthread_key_t` destructor
+  - New module: `py_thread_handler.erl` - Coordinator and per-thread handlers
+  - New C file: `py_thread_worker.c` - Thread worker pool management
+  - New test suite: `test/py_thread_callback_SUITE.erl`
+
 - **Reentrant Callbacks** - Python→Erlang→Python callback chains without deadlocks
   - Exception-based suspension mechanism interrupts Python execution cleanly
   - Callbacks execute in separate processes to prevent worker pool exhaustion

c_src/py_callback.c

@@ -540,11 +540,42 @@ static PyObject *ErlangFunction_New(PyObject *name) {
 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 like ThreadPoolExecutor (thread worker path).
+     */
     if (tl_current_worker == NULL || !tl_current_worker->has_callback_handler) {
-        PyErr_SetString(PyExc_RuntimeError,
-            "erlang.call() must be called from worker thread. "
-            "Use execute_async() for concurrent calls from other threads.");
-        return NULL;
+        /*
+         * Not an executor thread - try thread worker path.
+         * This enables ThreadPoolExecutor threads to call erlang.call().
+         */
+        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);

c_src/py_nif.c

@@ -113,6 +113,7 @@ ERL_NIF_TERM ATOM_SUSPENDED;
 #include "py_convert.c"
 #include "py_exec.c"
 #include "py_callback.c"
+#include "py_thread_worker.c"
 
 /* ============================================================================
  * Resource callbacks
@@ -388,6 +389,11 @@ static ERL_NIF_TERM nif_py_init(ErlNifEnv *env, int argc, const ERL_NIF_TERM arg
         return make_error(env, "executor_start_failed");
     }
 
+    /* Initialize thread worker system for ThreadPoolExecutor support */
+    if (thread_worker_init() < 0) {
+        /* Non-fatal - thread worker support just won't be available */
+    }
+
     return ATOM_OK;
 }
 
@@ -399,6 +405,9 @@ static ERL_NIF_TERM nif_finalize(ErlNifEnv *env, int argc, const ERL_NIF_TERM ar
         return ATOM_OK;
     }
 
+    /* Clean up thread worker system */
+    thread_worker_cleanup();
+
     /* Stop executors based on mode */
     switch (g_execution_mode) {
         case PY_MODE_FREE_THREADED:
@@ -1744,7 +1753,12 @@ static ErlNifFunc nif_funcs[] = {
 
     /* Execution mode info */
     {"execution_mode", 0, nif_execution_mode, 0},
-    {"num_executors", 0, nif_num_executors, 0}
+    {"num_executors", 0, nif_num_executors, 0},
+
+    /* Thread worker support (ThreadPoolExecutor) */
+    {"thread_worker_set_coordinator", 1, nif_thread_worker_set_coordinator, 0},
+    {"thread_worker_write", 2, nif_thread_worker_write, 0},
+    {"thread_worker_signal_ready", 1, nif_thread_worker_signal_ready, 0}
 };
 
 ERL_NIF_INIT(py_nif, nif_funcs, load, NULL, upgrade, unload)

c_src/py_nif.h

@@ -1151,4 +1151,55 @@ static void detect_execution_mode(void);
 
 /** @} */
 
+/* ============================================================================
+ * Thread Worker Functions (py_thread_worker.c)
+ * ============================================================================ */
+
+/**
+ * @defgroup thread_worker Thread Worker Support
+ * @brief Functions for ThreadPoolExecutor thread support
+ * @{
+ */
+
+/**
+ * @brief Initialize the thread worker system
+ *
+ * Creates pthread key for automatic cleanup on thread exit.
+ * Called during NIF initialization.
+ *
+ * @return 0 on success, -1 on failure
+ */
+static int thread_worker_init(void);
+
+/**
+ * @brief Clean up the thread worker system
+ *
+ * Releases all workers in the pool and destroys the pthread key.
+ * Called during NIF unload.
+ */
+static void thread_worker_cleanup(void);
+
+/**
+ * @brief Set the thread coordinator PID
+ *
+ * @param pid PID of the coordinator process
+ */
+static void thread_worker_set_coordinator(ErlNifPid pid);
+
+/**
+ * @brief Execute an erlang.call() from a spawned thread
+ *
+ * Called when a Python thread that is NOT an executor thread
+ * tries to call erlang.call().
+ *
+ * @param func_name Name of the Erlang function to call
+ * @param func_name_len Length of function name
+ * @param call_args Python tuple of arguments
+ * @return Python result object, or NULL with exception set
+ */
+static PyObject *thread_worker_call(const char *func_name, size_t func_name_len,
+                                    PyObject *call_args);
+
+/** @} */
+
 #endif /* PY_NIF_H */