Fix thread safety issues in C NIF codebase

benoitc · benoitc · commit f23a696c3112 · 2026-02-16T18:28:27.000+01:00
- Use pthread_once for async callback initialization to prevent
  race condition on g_async_callback_initialized flag

- Fix mutex held during Python calls in async_event_loop_thread by
  collecting completed futures under mutex, releasing it, then
  processing callbacks outside the mutex

- Add read_with_timeout() and read_length_prefixed_data() helpers
  to py_nif.h with select()-based timeout support

- Update all blocking pipe reads to use timeout (30s for callbacks,
  10s for handler spawn) to prevent indefinite hangs
diff --git a/c_src/py_callback.c b/c_src/py_callback.c
@@ -625,39 +625,36 @@ static PyObject *erlang_call_impl(PyObject *self, PyObject *args) {
             func_term,
             args_term);
 
-        uint32_t response_len = 0;
-        ssize_t n;
         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);
-        n = read(tl_current_worker->callback_pipe[0], &response_len, sizeof(response_len));
+        /* 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 (n != sizeof(response_len)) {
-            PyErr_SetString(PyExc_RuntimeError, "Failed to read callback response length");
+        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;
         }
-
-        response_data = enif_alloc(response_len);
-        if (response_data == NULL) {
+        if (read_result == -2) {
             PyErr_SetString(PyExc_MemoryError, "Failed to allocate response buffer");
             return NULL;
         }
 
-        Py_BEGIN_ALLOW_THREADS
-        n = read(tl_current_worker->callback_pipe[0], response_data, response_len);
-        Py_END_ALLOW_THREADS
-
-        if (n != (ssize_t)response_len) {
+        PyObject *result = parse_callback_response((unsigned char *)response_data, response_len);
+        if (response_data != NULL) {
             enif_free(response_data);
-            PyErr_SetString(PyExc_RuntimeError, "Failed to read callback response data");
-            return NULL;
         }
-
-        PyObject *result = parse_callback_response((unsigned char *)response_data, response_len);
-        enif_free(response_data);
         return result;
     }
 
@@ -723,19 +720,19 @@ extern bool g_has_thread_coordinator;
 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;
-static bool g_async_callback_initialized = false;
+
+/* 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;
 
 /**
- * Initialize async callback system.
- * Creates the response pipe and pending futures dict.
+ * Internal initialization function called by pthread_once.
+ * Thread-safe: only called once by pthread_once.
  */
-static int async_callback_init(void) {
-    if (g_async_callback_initialized) {
-        return 0;
-    }
-
+static void async_callback_init_impl(void) {
     if (pipe(g_async_callback_pipe) < 0) {
-        return -1;
+        g_async_callback_init_result = -1;
+        return;
     }
 
     /* Set the read end to non-blocking for asyncio compatibility */
@@ -750,11 +747,21 @@ static int async_callback_init(void) {
         close(g_async_callback_pipe[1]);
         g_async_callback_pipe[0] = -1;
         g_async_callback_pipe[1] = -1;
-        return -1;
+        g_async_callback_init_result = -1;
+        return;
     }
 
-    g_async_callback_initialized = true;
-    return 0;
+    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;
 }
 
 /**
@@ -893,11 +900,10 @@ static PyObject *get_async_callback_fd(PyObject *self, PyObject *args) {
     (void)self;
     (void)args;
 
-    if (!g_async_callback_initialized) {
-        if (async_callback_init() < 0) {
-            PyErr_SetString(PyExc_RuntimeError, "Failed to initialize async callback system");
-            return NULL;
-        }
+    /* 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]);
@@ -1328,40 +1334,75 @@ static void *async_event_loop_thread(void *arg) {
             PyErr_Clear();
         }
 
-        /* Check for completed futures (GIL held) */
+        /*
+         * 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 *prev = NULL;
         async_pending_t *p = worker->pending_head;
-        while (p != NULL) {
+        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);
 
-                    /* Future is complete - process it */
-                    async_future_callback(worker, p);
+        /* Phase 2: Process completed callbacks outside mutex (no deadlock risk) */
+        for (int i = 0; i < num_completed; i++) {
+            async_future_callback(worker, completed[i]);
+        }
 
-                    /* Remove from list */
-                    Py_DECREF(p->future);
-                    if (prev == NULL) {
-                        worker->pending_head = p->next;
-                    } else {
-                        prev->next = p->next;
-                    }
-                    if (p == worker->pending_tail) {
-                        worker->pending_tail = prev;
+        /* 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;
                     }
-                    async_pending_t *to_free = p;
+                    prev = p;
                     p = p->next;
-                    enif_free(to_free);
-                    continue;
                 }
-                Py_XDECREF(done);
+
+                /* Clean up */
+                Py_DECREF(to_remove->future);
+                enif_free(to_remove);
             }
-            prev = p;
-            p = p->next;
+            pthread_mutex_unlock(&worker->queue_mutex);
         }
-        pthread_mutex_unlock(&worker->queue_mutex);
     }
 
     /* Stop and close the event loop */
diff --git a/c_src/py_nif.h b/c_src/py_nif.h
@@ -98,6 +98,8 @@
 /** @brief Flag indicating dlopen with RTLD_GLOBAL is needed for Python extensions */
 #define NEED_DLOPEN_GLOBAL 1
 #endif
+
+#include <sys/select.h>
 /** @} */
 
 /* ============================================================================
@@ -871,6 +873,86 @@ static char *binary_to_string(const ErlNifBinary *bin) {
     return str;
 }
 
+/**
+ * @brief Read from a file descriptor with optional timeout
+ *
+ * Uses select() to implement a timeout on blocking read operations.
+ * This prevents indefinite blocking on pipe reads.
+ *
+ * @param fd         File descriptor to read from
+ * @param buf        Buffer to read into
+ * @param count      Number of bytes to read
+ * @param timeout_ms Timeout in milliseconds (0 = no timeout, wait indefinitely)
+ *
+ * @return Number of bytes read on success, 0 on timeout, -1 on error
+ *
+ * @note On timeout, errno is set to ETIMEDOUT
+ */
+static ssize_t read_with_timeout(int fd, void *buf, size_t count, int timeout_ms) {
+    if (timeout_ms > 0) {
+        struct timeval tv;
+        tv.tv_sec = timeout_ms / 1000;
+        tv.tv_usec = (timeout_ms % 1000) * 1000;
+
+        fd_set fds;
+        FD_ZERO(&fds);
+        FD_SET(fd, &fds);
+
+        int ret = select(fd + 1, &fds, NULL, NULL, &tv);
+        if (ret < 0) {
+            return -1;  /* select error */
+        }
+        if (ret == 0) {
+            errno = ETIMEDOUT;
+            return 0;  /* timeout */
+        }
+    }
+
+    return read(fd, buf, count);
+}
+
+/**
+ * @brief Read length-prefixed data from a file descriptor
+ *
+ * Reads a 4-byte length prefix followed by the data payload.
+ * Uses read_with_timeout for optional timeout support.
+ *
+ * @param fd         File descriptor to read from
+ * @param data_out   Pointer to store allocated data buffer (caller must free with enif_free)
+ * @param len_out    Pointer to store data length
+ * @param timeout_ms Timeout in milliseconds (0 = no timeout)
+ *
+ * @return 0 on success, -1 on read error/timeout, -2 on allocation failure
+ *
+ * @note On success with len > 0, caller must call enif_free(*data_out)
+ */
+static int read_length_prefixed_data(int fd, char **data_out, uint32_t *len_out, int timeout_ms) {
+    uint32_t len;
+    ssize_t n = read_with_timeout(fd, &len, sizeof(len), timeout_ms);
+    if (n != sizeof(len)) {
+        return -1;
+    }
+
+    *data_out = NULL;
+    *len_out = len;
+
+    if (len > 0) {
+        *data_out = enif_alloc(len);
+        if (*data_out == NULL) {
+            return -2;
+        }
+
+        n = read_with_timeout(fd, *data_out, len, timeout_ms);
+        if (n != (ssize_t)len) {
+            enif_free(*data_out);
+            *data_out = NULL;
+            return -1;
+        }
+    }
+
+    return 0;
+}
+
 /** @} */
 
 /* ============================================================================
diff --git a/c_src/py_thread_worker.c b/c_src/py_thread_worker.c
@@ -332,9 +332,14 @@ static int thread_worker_spawn_handler(thread_worker_t *tw) {
     }
     enif_free_env(msg_env);
 
-    /* Read handler PID from pipe (Erlang writes it after spawning) */
+    /* Read handler PID from pipe (Erlang writes it after spawning)
+     * Use 10 second timeout for handler spawn */
     uint32_t response_len = 0;
-    ssize_t n = read(tw->response_pipe[0], &response_len, sizeof(response_len));
+    ssize_t n = read_with_timeout(tw->response_pipe[0], &response_len, sizeof(response_len), 10000);
+    if (n <= 0) {
+        /* Timeout or error - handler spawn failed */
+        return -1;
+    }
     if (n != sizeof(response_len) || response_len == 0) {
         /* Handler spawned successfully - pid info is in the length as a signal */
         tw->has_handler = true;
@@ -420,9 +425,9 @@ static PyObject *thread_worker_call(const char *func_name, size_t func_name_len,
         func_term,
         args_term);
 
-    ssize_t n;
-    uint32_t response_len = 0;
     char *response_data = NULL;
+    uint32_t response_len = 0;
+    int read_result;
 
     /* Send message to coordinator (can be done with GIL held) */
     if (!enif_send(NULL, &g_thread_coordinator_pid, msg_env, msg)) {
@@ -434,36 +439,29 @@ static PyObject *thread_worker_call(const char *func_name, size_t func_name_len,
 
     /* Release GIL while waiting for response */
     Py_BEGIN_ALLOW_THREADS
-
-    /* Read response from pipe */
-    n = read(tw->response_pipe[0], &response_len, sizeof(response_len));
-
+    /* Use 30 second timeout to prevent indefinite blocking */
+    read_result = read_length_prefixed_data(
+        tw->response_pipe[0], &response_data, &response_len, 30000);
     Py_END_ALLOW_THREADS
 
-    if (n != sizeof(response_len)) {
-        PyErr_SetString(PyExc_RuntimeError, "Failed to read callback response length");
+    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;
     }
-
-    response_data = enif_alloc(response_len);
-    if (response_data == NULL) {
+    if (read_result == -2) {
         PyErr_SetString(PyExc_MemoryError, "Failed to allocate response buffer");
         return NULL;
     }
 
-    Py_BEGIN_ALLOW_THREADS
-    n = read(tw->response_pipe[0], response_data, response_len);
-    Py_END_ALLOW_THREADS
-
-    if (n != (ssize_t)response_len) {
-        enif_free(response_data);
-        PyErr_SetString(PyExc_RuntimeError, "Failed to read callback response data");
-        return NULL;
-    }
-
     /* Parse response using existing function */
     PyObject *result = parse_callback_response((unsigned char *)response_data, response_len);
-    enif_free(response_data);
+    if (response_data != NULL) {
+        enif_free(response_data);
+    }
 
     return result;
 }
